Python aiohttp中间件示例

你的名字 2023-02-13 05:53 16阅读 0赞

`aiohttp`

# 获取表单
    data = await self.request.post()
    
    # 获取url /{user_id}/ 中的参数
    user_id = self.request.match_info.get('user_id')
    
    # 获取Json
    data = await self.request.json()
    
    # 获取get参数
    name = self.request.query.get("name")

`aiohttp 中间件`

from aiohttp import web
    from functools import wraps
    
    @web.middleware
    async def middleware1(request, handler):
        print('Middleware 1 called')
        response = await handler(request)
        print('Middleware 1 finished')
        return response
    
    @web.middleware
    async def middleware2(request, handler):
        print('Middleware 2 called')
        response = await handler(request)
        print('Middleware 2 finished')
        return response
    
    
    async def logger_factory_003(app,handler):
        async def logger1_handler(request):
    
            print('i am logger factory 003')
            return await handler(request)
        return logger1_handler
    
    
    def middleware_004(func):
        @wraps(func)
        async def inner(request, *args, **kwargs):
            print(">>>>>>>>> 004")
            resp =  await func(request, *args, **kwargs)
            print("<<<<<<<<< 004")
            return resp
        return inner
    
    
    # 004是这有这个有用
    @middleware_004
    async def test(request):
        print('Handler function called')
        return web.Response(text="Hello")
    
    # 这里注册的中间件是全局的
    app = web.Application(middlewares=[middleware1, middleware2,logger_factory_003])
    app.router.add_get('/', test)
    web.run_app(app,host="0.0.0.0",port="8888")

Middleware 1 called
    Middleware 2 called
    i am logger factory 003
    >>>>>>>>>  004
    Handler function called
    <<<<<<<<<  004
    Middleware 2 finished
    Middleware 1 finished

如果是类View, 全局注册的中间件不需要更改，只有004号直接作用到view上的需要改.

如我下面摘录的代码

import jwt
    from aiohttp import web
    from functools import wraps
    from settings import JWT_SECRET_KEY
    from dao.utils import json_dumps
    
    def jwt_token_required(func):
        """This function applies only to class views."""
        # jwt.exceptions.DecodeError token 格式不对
        # jwt.exceptions.InvalidSignatureError 秘钥key不对
        # jwt.exceptions.InvalidAlgorithmErro 算法不一致
        # jwt.exceptions.ExpiredSignatureError token过期
    
        @wraps(func)
        async def inner(cls, *args, **kwargs):
            jwt_token = cls.request.query.get('token')
            if not jwt_token:
                data = { "code": 10005, "msg": "缺少token"}
                return web.json_response(data=data, dumps=json_dumps)
            try:
                jwt_payload = jwt.decode(jwt_token, JWT_SECRET_KEY, algorithms=['HS256'])
                cls.request.app["jwt_token"] = jwt_payload
                return await func(cls, *args, **kwargs)  # 正常调用并返回
            except (
                jwt.exceptions.DecodeError,
                jwt.exceptions.InvalidSignatureError,
                jwt.exceptions.InvalidAlgorithmError
                ) as e:
    
                data = { "code": 10006, "msg": "token不正确"}
                return web.json_response(data=data, dumps=json_dumps)
            except jwt.exceptions.ExpiredSignatureError as e:
                data = { "code": 10007, "msg": "token已过期"}
                return web.json_response(data=data, dumps=json_dumps)
    
        return inner