response.py 842 B

12345678910111213141516171819202122232425262728293031
  1. # app/utils/response.py
  2. from app.schemas.common import ApiResponse
  3. from typing import Any
  4. from pydantic import BaseModel
  5. def _to_serializable(data: Any):
  6. """自动将 ORM / Pydantic / list 转换为可序列化对象"""
  7. if isinstance(data, BaseModel):
  8. return data.dict()
  9. if hasattr(data, "__table__"): # SQLAlchemy ORM
  10. return {
  11. c.name: getattr(data, c.name)
  12. for c in data.__table__.columns
  13. }
  14. if isinstance(data, list):
  15. return [_to_serializable(i) for i in data]
  16. return data
  17. def success(data=None, message: str = "success"):
  18. return ApiResponse(
  19. code=0,
  20. message=message,
  21. data=_to_serializable(data)
  22. )
  23. def fail(message: str, code: int = 1):
  24. return ApiResponse(
  25. code=code,
  26. message=message,
  27. data=None
  28. )