Python从入门到入土-web应用开发

2022-11-28 15:48:59 浏览数 (1)

CSDN话题挑战赛第2期 参赛话题：学习笔记

web开发基础知识

错误码处理

代码语言：javascript复制
# 编写一个错误码枚举，支持转换成字符串格式方法： internal_ret_2_http

# -*- coding: UTF-8 -*-
from enum import Enum

class ErrorCode(Enum):
    SUCCESS = 0
    FAILED = 1
    NOT_FOUND = 2
    ALREADY_EXIST = 3
    INVALID_PARAMETERS = 4

    @staticmethod
    def internal_ret_2_http(ret):
        ret['err'] = ret['err'].name.lower()

if __name__ == '__main__':
    ret = {'err': ErrorCode.NOT_FOUND}
    ErrorCode.internal_ret_2_http(ret)
    assert ret['err'] == 'not_found'

状态存储

代码语言：javascript复制
# 使用dict数据结构，实现创建、删除、查询总数存储接口，创建/删除同一个资源后，总是应该为0

# -*- coding: UTF-8 -*-
from error_code import ErrorCode
import logging
logger = logging.getLogger(__name__)

class Store:
    def __init__(self, config) -> None:
        self.config = config
        self.records = {}

    def __where(self, key, condition):
        if condition is None:
            return True
        return self.records.get(key) == condition

    def create(self, key, value):
        if self.records.get(key) is None:
            self.records[key] = value
            return {'err': ErrorCode.SUCCESS}
        else:
            return {'err': ErrorCode.ALREADY_EXIST}

    def update(self, key, value, condition=None):
        if self.__where(key, condition):
            self.records[key] = value
            return {'err': ErrorCode.SUCCESS}
        else:
            return {'err': ErrorCode.NOT_FOUND}

    def remove(self, key, condition=None):
        if self.__where(key, condition):
            del self.records[key]
            return {'err': ErrorCode.SUCCESS}
        else:
            return {'err': ErrorCode.NOT_FOUND}

    def count(self):
        return {'err': ErrorCode.SUCCESS, 'result': len(self.records.keys())}

    # 请实现一个根据key 和条件 condition 查找的方法
    def fetch(self, key, condition=None):
        if self.__where(key, condition):
            result = self.records.get(key)
            if result is None:
                return {'err': ErrorCode.NOT_FOUND}
            else:
                return {'err': ErrorCode.SUCCESS, 'result': [result]}
        else:
            return {'err': ErrorCode.NOT_FOUND}


if __name__ == '__main__':
    store = Store({})
    ret = store.create("test",  100)
    assert ret['err'] == ErrorCode.SUCCESS

    ret = store.remove("test")
    assert ret['err'] == ErrorCode.SUCCESS
    ret = store.count()
    assert ret['err'] == ErrorCode.SUCCESS
    assert ret['result'] == 0

路由器

代码语言：javascript复制
#
# 下面是一个 HTTP 状态代码的定义：
#
# 2xx：成功:
#
# 200 正常，请求已完成。
# 201 正常，紧接POST命令。
# 202 正常，已接受用于处理，但处理尚未完成。
# 203 正常，部分信息—返回的信息只是一部分。
# 204 正常，无响应—已接收请求，但不存在要回送的信息。

# 3xx：重定向:
#
# 301 已移动，请求的数据具有新的位置且更改是永久的。
# 302 已找到，请求的数据临时具有不同 URI。
# 303 请参阅其它，可在另一 URI 下找到对请求的响应，且应使用 GET 方法检索此响应。
# 304 未修改，未按预期修改文档。
# 305 使用代理，必须通过位置字段中提供的代理来访问请求的资源。
# 306 未使用，不再使用，保留此代码以便将来使用。

# 4xx：客户机中出现的错误:
#
# 400 错误请求，请求中有语法问题，或不能满足请求。
# 401 未授权，未授权客户机访问数据。
# 402 需要付款，表示计费系统已有效。
# 403 禁止，即使有授权也不需要访问。
# 404 找不到，服务器找不到给定的资源；文档不存在。
# 407 代理认证请求，客户机首先必须使用代理认证自身。
# 415 介质类型不受支持，服务器拒绝服务请求，因为不支持请求实体的格式。

# 5xx：服务器中出现的错误:
#
# 500 内部错误，因为意外情况，服务器不能完成请求。
# 501 未执行，服务器不支持请求的工具。
# 502 错误网关，服务器接收到来自上游服务器的无效响应。
# 503 无法获得服务，由于临时过载或维护，服务器无法处理请求。
# 编写一个路由服务，支持注入路由配置，正确处理请求参数

# -*- coding: UTF-8 -*-
from error_code import ErrorCode
import json
import logging
import traceback
logger = logging.getLogger(__name__)

class Router:
    def __init__(self, routes) -> None:
        self.routes = routes

    # 请实现路由逻辑，返回符合语义的HTTP状态码
    def dispatch(self, http_request_str):
        req_path = None
        req = None
        try:
            http_request = json.loads(http_request_str)
            req_path = http_request.get('path')
            req = http_request.get('data')
        except Exception as e:
            logger.error(f"parse data exception:{str(e)}")
            return {'err': ErrorCode.INVALID_PARAMETERS}, 400

        if req_path is None or self.routes.get(req_path) is None:
            return {'err': ErrorCode.NOT_FOUND}, 404

        try:
            handler = self.routes.get(req_path)
            return handler(req), 200
        except Exception as e:
            logger.error(f"route to '{req_path}' exception:{str(e)}")
            logger.error(traceback.format_exc())
            return {'err': ErrorCode.FAILED}, 500

if __name__ == '__main__':
    # 注册路由
    router = Router({
        '/': lambda req: print(f'goto home:{str(req)}')
    })

    # 分发路由
    router.dispatch(json.dumps({
        'path': '/',
        'data': ["Hello", "World!"]
    }))

参数校验

代码语言：javascript复制
# 使用 jsonschema 校验参数，key:字符串类型, value:数字, condition: 数字

# -*- coding: UTF-8 -*-
from error_code import ErrorCode
from jsonschema import validate
import json
import logging
import traceback
logger = logging.getLogger(__name__)

class KeyValueValidator:
    def __init__(self) -> None:
        pass

    '''使用jsonschema校验参数
    req: 请求参数
    required： 必须要有的字段
    '''
    def validate(self, req, required):
        # 正确配置jsonschema
        schema = {
            "type": "object",
            "properties": {
                "key": {"type": "string"},
                "value": {"type": "number"},
                "condition": {"type": "number"},
            },
            "required": required
        }

        try:
            validate(instance=req, schema=schema)
            return {
                'err': ErrorCode.SUCCESS
            }
        except Exception as e:
            logger.error(f"validate exception:{str(e)}")
            logger.error(traceback.format_exc())
            return {
                'err': ErrorCode.INVALID_PARAMETERS
            }


if __name__ == '__main__':
    v = KeyValueValidator()

    ret = v.validate({'key': "test", 'value': 100,'condition':666}, ['key', 'value'])
    assert ret['err'] == ErrorCode.SUCCESS
    ret = v.validate({'key': "test"}, ['key', 'value'])
    assert ret['err'] == ErrorCode.INVALID_PARAMETERS

Python Web 服务模拟器

代码语言：javascript复制
# 综合使用前2节的ErrorCode、Router两个类，模拟一个 Web 服务，支持：
#
# 创建资源
# 删除资源
# 统计资源个数
# 那么，先创建一个资源，接着删除同一个资源，最后统计资源个数，总数应该为0

# -*- coding: UTF-8 -*-
from error_code import ErrorCode
from store import Store
from router import Router
from validator import KeyValueValidator
import json
import logging
import traceback
logger = logging.getLogger(__name__)

class App:
    def __init__(self) -> None:
        self.store = Store({})
        self.validator = KeyValueValidator()

        # 创建一个路由器，支持3个目标API
        self.router = Router({
            '/': lambda req: self.__home(req),
            '/kv/create': lambda req: self.__create(req),
            '/kv/remove': lambda req: self.__remove(req),
            '/kv/count': lambda req: self.__count(req),
        })


    # 请正确实现 post 方法，接受 API 请求
    def post(self, path, data):
        '''HTTP POST方法模拟实现
        path: 请求路径
        data: 请求数据，使用JSON模拟
        '''
        http_request = {
            'path': path,
            'data': data
        }
        ret, status_code = self.router.dispatch(json.dumps(http_request))
        ErrorCode.internal_ret_2_http(ret)
        resp = ""
        try:
            resp = json.dumps(ret)
        except Exception as e:
            logger.error("parse resp exception:%s", str(e))
            logger.error(traceback.format_exc())
        return resp, status_code

    def __home(self, req):
        # 首页
        return {'err': ErrorCode.SUCCESS, 'result': "Welcome!"}

    def __create(self, req):
        # 创建资源
        ret = self.validator.validate(req, ['key', 'value'])
        if ret['err'] != ErrorCode.SUCCESS:
            return ret

        return self.store.create(req['key'], req['value'])

    def __remove(self, req):
        # 移除资源
        ret = self.validator.validate(req, ['key', 'condition'])
        if ret['err'] != ErrorCode.SUCCESS:
            return ret

        return self.store.remove(req['key'], req['condition'])

    def __count(self, req):
        # 统计资源个数
        return self.store.count()


if __name__ == '__main__':
    app = App()

    resp, status = app.post('/kv/create', {
        'key': 'test',
        'value': 1000,
    })

    resp, status = app.post('/kv/remove', {
        'key': 'test',
        'condition': 1000,
    })

    resp, status = app.post('/kv/count', {})
    assert status == 200
    assert json.loads(resp)['err'] == 'success'
    assert json.loads(resp)['result'] == 0

Python Django 框架

Django是高水准的Python编程语言驱动的一个开源模型．视图，控制器风格的Web应用程序框架，它起源于开源社区。使用这种架构，程序员可以方便、快捷地创建高品质、易维护、数据库驱动的应用程序。 这也正是OpenStack的Horizon组件采用这种架构进行设计的主要原因。另外，在Dj ango框架中，还包含许多功能强大的第三方插件，使得Django具有较强的可扩展性 。 Django 项目源自一个在线新闻 Web 站点，于 2005 年以开源的形式被释放出来。Django 框架的核心组件有：

  1. 用于创建模型的对象关系映射；
  2. 为最终用户设计较好的管理界面；
  3. URL 设计；
  4. 设计者友好的模板语言；
  5. 缓存系统。

Django已经成为web开发者的首选框架，是一个遵循 MVC 设计模式的框架。MVC是Model、View、Controller三个单词的简写，分别代表模型、视图、控制器。Django其实也是一个MTV 的设计模式。 MTV是Model、Template、View三个单词的简写，分别代表模型、模版、视图 [4]  。但是在Django中，控制器接受用户输入的部分由框架自行处理， 所以 Django 里更关注的是模型（Model）、模板(Template)和视图（Views），称为 MTV模式。

代码语言：javascript复制
if __name__ == '__main__':

    # 在此添加创建django项目和初始化目录的命令
    create_django = 'django-admin startproject projectName'
    init_django = 'python manage.py startapp projectApp'

    installs = [
        "安装：pip install django",
        f"创建django项目命令：{create_django}",
        "进入目录：cd projectName",
        f"初始化django项目命令：{init_django}",
        "进一步查看教程：https://docs.djangoproject.com/zh-hans/3.2/"
    ]
    for step in installs:
        print(step)

Python tornado 框架

tornado 服务，极简路由. Tornado 是一个基于 Python 的 Web 服务框架和异步网络库。 最早开发于 FriendFeed 公司，通过利用非阻塞网络 I/O, Tornado 可以承载成千上万的活动连接, 完美的实现了长连接, WebSockets, 和其他对于每一位用户来说需要长连接的程序。

安装：

代码语言：javascript复制
pip install tornado 【安装最新稳定版】
 
pip install tornado==version 【指定版本安装】
代码语言：javascript复制
#
# tornado 服务，极简路由

# -*- coding: UTF-8 -*-
import tornado.web
import tornado.ioloop

class IndexHandler(tornado.web.RequestHandler):
    """主路由处理类"""

    def get(self):
        """对应http的get请求方式"""
        # 实现Tornado get 方法
        self.write("Hello Tornado!")

if __name__ == "__main__":
    app = tornado.web.Application([
        (r"/", IndexHandler),
    ])
    app.listen(8000)

    print("* Tornado Web Server 已在 8000 端口启动。")
    print("* 请在浏览器里输入 127.0.0.1:8000")
    tornado.ioloop.IOLoop.current().start()

Flask简单使用

代码语言：javascript复制
# Flask简单使用
#
# 用flask启动web服务，响应根页面HTTP GET请求：'/'，返回"<p>Hello World!</p>"

import sys
from flask import Flask
from flask_cors import CORS
from gevent import pywsgi, monkey

monkey.patch_all()

app = Flask(
    __name__,
    static_folder='web',
    static_url_path=''
)

def after_request(resp):
    resp.headers['Access-Control-Allow-Origin'] = '*'
    return resp

app.after_request(after_request)
app.config['JSON_AS_ASCII'] = False
CORS(app, supports_credentials=True)

# 在此添加Flask API响应
@app.route('/', methods=['GET'])
def home():
    return "<p>Hello World!</p>"

if __name__ == '__main__':
    host = '127.0.0.1'
    port = 1024
    print('@启动服务...')
    print("@本地调试：http://{}:{}".format(host, port))
    if len(sys.argv) > 1 and sys.argv[1] == 'debug':
        app.run(host=host, port=port)
    else:
        server = pywsgi.WSGIServer((host, port), app)
        server.serve_forever()
django python 开源 mvc

0 人点赞