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

水深无声 2023-09-27 10:59 146阅读 0赞

[CSDN话题挑战赛第2期][CSDN_2]  
参赛话题：[学习笔记][Link 1]

### web开发基础知识 ###

#### 错误码处理 ####

# 编写一个错误码枚举，支持转换成字符串格式方法： 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'

#### 状态存储 ####

# 使用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

#### 路由器 ####

#
    # 下面是一个 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!"]
        }))

#### 参数校验 ####

# 使用 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 服务模拟器 ####

# 综合使用前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模式。

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, 和其他对于每一位用户来说需要长连接的程序。

安装：

pip install tornado 【安装最新稳定版】
     
    pip install tornado==version 【指定版本安装】

#
    # 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简单使用 ###

# 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()

[CSDN_2]: https://marketing.csdn.net/p/7b6697fd9dd3795a268d1a6f2fe75012
[Link 1]: https://activity.csdn.net/creatActivity?id=10213