response.py 1.2 KB

1234567891011121314151617181920212223242526272829303132333435363738394041424344454647484950515253
  1. """DMS统一响应与序列化工具。"""
  2. from __future__ import annotations
  3. from typing import Any
  4. from flask import Response, jsonify
  5. from dms.common.request_context import REQUEST_ID_HEADER, get_request_id
  6. def success_response(
  7. data: Any = None,
  8. *,
  9. message: str = "success",
  10. status: int = 200,
  11. ) -> tuple[Response, int]:
  12. request_id = get_request_id()
  13. response = jsonify(
  14. {
  15. "code": "OK",
  16. "message": message,
  17. "data": {} if data is None else data,
  18. "requestId": request_id,
  19. }
  20. )
  21. response.headers[REQUEST_ID_HEADER] = request_id
  22. return response, status
  23. def error_response(
  24. code: str,
  25. message: str,
  26. *,
  27. details: Any = None,
  28. status: int = 500,
  29. ) -> tuple[Response, int]:
  30. request_id = get_request_id()
  31. response = jsonify(
  32. {
  33. "code": code,
  34. "message": message,
  35. "details": details,
  36. "requestId": request_id,
  37. }
  38. )
  39. response.headers[REQUEST_ID_HEADER] = request_id
  40. return response, status
  41. def serialize_id(value: int | None) -> str | None:
  42. """将数据库BIGINT安全序列化为十进制字符串。"""
  43. return None if value is None else str(value)