Fix use request.form() in middleware (#260)

* Fix request.form() in middleware

* resume global exception handling

* update the order in which data is processed

* Add demo interface for uploading files

* fix path params

Author: wu-clan

backend/app/api/v1/mixed/tests.py

@@ -1,20 +1,29 @@
 #!/usr/bin/env python3
 # -*- coding: utf-8 -*-
-from fastapi import APIRouter
+from typing import Annotated
+
+from fastapi import APIRouter, File, Form, UploadFile
 
 from backend.app.common.response.response_schema import response_base
 from backend.app.tasks import task_demo_async
 
 router = APIRouter(prefix='/tests')
 
 
-@router.post('/send', summary='测试异步任务')
-async def task_send():
-    result = task_demo_async.delay()
-    return await response_base.success(data=result.id)
-
-
 @router.post('/send', summary='异步任务演示')
 async def send_task():
     result = task_demo_async.delay()
     return await response_base.success(data=result.id)
+
+
+@router.post('/files', summary='上传文件演示')
+async def create_file(
+    file: Annotated[bytes, File()],
+    fileb: Annotated[UploadFile, File()],
+    token: Annotated[str, Form()],
+):
+    return {
+        'file_size': len(file),
+        'token': token,
+        'fileb_content_type': fileb.content_type,
+    }

backend/app/middleware/opera_log_middleware.py

@@ -1,12 +1,11 @@
 #!/usr/bin/env python3
 # -*- coding: utf-8 -*-
-from typing import Any, AsyncGenerator
-
 from asgiref.sync import sync_to_async
+from fastapi import Response
 from starlette.background import BackgroundTask
 from starlette.datastructures import UploadFile
+from starlette.middleware.base import BaseHTTPMiddleware
 from starlette.requests import Request
-from starlette.types import ASGIApp, Receive, Scope, Send
 
 from backend.app.common.enums import OperaLogCipherType
 from backend.app.common.log import log
@@ -18,26 +17,16 @@
 from backend.app.utils.timezone import timezone
 
 
-class OperaLogMiddleware:
+class OperaLogMiddleware(BaseHTTPMiddleware):
     """操作日志中间件"""
 
-    def __init__(self, app: ASGIApp):
-        self.app = app
-
-    async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None:
-        if scope['type'] != 'http':
-            await self.app(scope, receive, send)
-            return
-
-        request = Request(scope=scope, receive=receive)
-
+    async def dispatch(self, request: Request, call_next) -> Response:
         # 排除记录白名单
         path = request.url.path
         if path in settings.OPERA_LOG_EXCLUDE or not path.startswith(f'{settings.API_V1_STR}'):
-            await self.app(scope, receive, send)
-            return
+            return await call_next(request)
 
-        # 请求信息解析
+        # 请求解析
         user_agent, device, os, browser = await parse_user_agent_info(request)
         ip, country, region, city = await parse_ip_info(request)
         try:
@@ -46,10 +35,10 @@ async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None:
         except AttributeError:
             username = None
         method = request.method
-        args = await self.get_request_args(request)
         router = request.scope.get('route')
         summary = getattr(router, 'summary', None) or ''
-        args.update(request.path_params)
+        args = await self.get_request_args(request)
+        args = await self.desensitization(args)
 
         # 设置附加请求信息
         request.state.ip = ip
@@ -63,13 +52,10 @@ async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None:
 
         # 执行请求
         start_time = timezone.now()
-        code, msg, status, err = await self.execute_request(request, send)
+        code, msg, status, err, response = await self.execute_request(request, call_next)
         end_time = timezone.now()
         cost_time = (end_time - start_time).total_seconds() * 1000.0
 
-        # 脱敏处理
-        args = await self.desensitization(args)
-
         # 日志创建
         opera_log_in = CreateOperaLog(
             username=username,
@@ -98,19 +84,15 @@ async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None:
         if err:
             raise err from None
 
-    async def execute_request(self, request: Request, send: Send) -> tuple:
-        err: Any = None
+        return response
+
+    async def execute_request(self, request: Request, call_next) -> tuple:
+        """执行请求"""
+        err = None
+        response = None
         try:
-            # 详见 https://github.com/tiangolo/fastapi/discussions/8385#discussioncomment-6117967
-            async def wrapped_rcv_gen() -> AsyncGenerator:
-                async for _ in request.stream():
-                    yield {'type': 'http.request', 'body': await request.body()}
-                    async for message in request.receive:  # type: ignore
-                        yield message
-
-            wrapped_rcv = wrapped_rcv_gen().__anext__
-            await self.app(request.scope, wrapped_rcv, send)
-            code, msg, status = await self.exception_middleware_handler(request)
+            response = await call_next(request)
+            code, msg, status = await self.request_exception_handler(request)
         except Exception as e:
             log.exception(e)
             # code 处理包含 SQLAlchemy 和 Pydantic
@@ -119,12 +101,12 @@ async def wrapped_rcv_gen() -> AsyncGenerator:
             status = 0
             err = e
 
-        return str(code), msg, status, err
+        return str(code), msg, status, err, response
 
     @staticmethod
     @sync_to_async
-    def exception_middleware_handler(request: Request) -> tuple:
-        # 预置响应信息
+    def request_exception_handler(request: Request) -> tuple:
+        """请求异常处理器"""
         code = 200
         msg = 'Success'
         status = 1
@@ -148,12 +130,16 @@ def exception_middleware_handler(request: Request) -> tuple:
 
     @staticmethod
     async def get_request_args(request: Request) -> dict:
+        """获取请求参数"""
         args = dict(request.query_params)
+        args.update(request.path_params)
+        # Tip: .body() 必须在 .form() 之前获取
+        # https://github.com/encode/starlette/discussions/1933
+        body_data = await request.body()
         form_data = await request.form()
         if len(form_data) > 0:
             args.update({k: v.filename if isinstance(v, UploadFile) else v for k, v in form_data.items()})
         else:
-            body_data = await request.body()
             if body_data:
                 json_data = await request.json()
                 if not isinstance(json_data, dict):
@@ -168,7 +154,15 @@ async def get_request_args(request: Request) -> dict:
     @staticmethod
     @sync_to_async
     def desensitization(args: dict) -> dict | None:
-        if len(args) > 0:
+        """
+        脱敏处理
+
+        :param args:
+        :return:
+        """
+        if not args:
+            args = None
+        else:
             match settings.OPERA_LOG_ENCRYPT:
                 case OperaLogCipherType.aes:
                     for key in args.keys():
@@ -188,4 +182,4 @@ def desensitization(args: dict) -> dict | None:
                     for key in args.keys():
                         if key in settings.OPERA_LOG_ENCRYPT_INCLUDE:
                             args[key] = '******'
-        return args if len(args) > 0 else None
+        return args

requirements.txt

@@ -10,7 +10,7 @@ celery==5.3.6
 cryptography==41.0.7
 email-validator==2.0.0
 fast-captcha==0.2.1
-fastapi==0.105.0
+fastapi==0.108.0
 fastapi-limiter==0.1.5
 fastapi-pagination==0.12.13
 gunicorn==21.2.0
@@ -33,7 +33,6 @@ pytz==2023.3
 redis[hiredis]==4.5.5
 ruff==0.1.8
 SQLAlchemy==2.0.23
-starlette==0.27.0
 supervisor==4.2.5
 user-agents==2.2.0
 uvicorn[standard]==0.24.0