admin_request_json.py 1.1 KB

1234567891011121314151617181920212223242526272829303132333435
  1. """运营 Web(/web/admin)JSON 请求体解析。
  2. 与 ApiCryptoMiddleware 配合:中间件未改写 body 时,在路由内解密 `{ enc, ts, nonce, edata }`
  3. 并剥 `{ data: ... }`,避免 FastAPI 声明式 Body 在校验阶段只看到密文信封而 422。
  4. """
  5. from __future__ import annotations
  6. from typing import Any, Optional, Tuple
  7. from fastapi import Request
  8. from app.core.request_payload import resolve_request_payload
  9. async def parse_admin_request_json(
  10. request: Request,
  11. current_user: dict,
  12. *,
  13. source: str,
  14. ) -> Tuple[Any, Optional[str]]:
  15. """
  16. 读取 JSON body;dict 时走 resolve_request_payload(明文 / 密文 / data 包装)。
  17. Returns:
  18. (payload, None) 成功;
  19. (_, err_msg) 密文解密失败(此时 payload 为原始 raw,调用方可忽略)。
  20. """
  21. raw: Any = await request.json()
  22. if isinstance(raw, dict):
  23. resolved = resolve_request_payload(raw, current_user, source=source)
  24. if resolved == {} and int(raw.get("enc", 0) or 0) == 1:
  25. return raw, "密文解密失败"
  26. return resolved, None
  27. return raw, None