response.py 2.0 KB

12345678910111213141516171819202122232425262728293031323334353637383940414243444546474849505152535455565758596061626364656667
  1. """
  2. 智能股票分析助手 — 统一响应格式模块
  3. 定义标准API响应结构,确保前后端数据格式一致。
  4. """
  5. from typing import Any, Optional
  6. from pydantic import BaseModel
  7. class APIResponse(BaseModel):
  8. """统一API响应格式"""
  9. code: int = 0 # 状态码:0=成功,非0=错误
  10. message: str = "success" # 提示信息
  11. data: Optional[Any] = None # 响应数据
  12. class Config:
  13. json_schema_extra = {
  14. "example": {
  15. "code": 0,
  16. "message": "success",
  17. "data": {"stock_code": "600519", "price": 1700.00},
  18. }
  19. }
  20. class PageResponse(BaseModel):
  21. """分页API响应格式"""
  22. code: int = 0
  23. message: str = "success"
  24. data: Optional[Any] = None
  25. pagination: Optional[dict] = None # 分页信息 {page, page_size, total, total_pages}
  26. class Config:
  27. json_schema_extra = {
  28. "example": {
  29. "code": 0,
  30. "message": "success",
  31. "data": [],
  32. "pagination": {"page": 1, "page_size": 20, "total": 100, "total_pages": 5},
  33. }
  34. }
  35. def success_response(data: Any = None, message: str = "success", meta: Any = None) -> dict:
  36. """快速构建成功响应;meta 用于妙想缓存/额度提示等扩展字段"""
  37. out: dict = {"code": 0, "message": message, "data": data}
  38. if meta is not None:
  39. out["meta"] = meta
  40. return out
  41. def error_response(code: int, message: str, data: Any = None) -> dict:
  42. """快速构建错误响应"""
  43. return APIResponse(code=code, message=message, data=data).model_dump()
  44. def page_response(data: Any, page: int, page_size: int, total: int) -> dict:
  45. """快速构建分页响应"""
  46. total_pages = (total + page_size - 1) // page_size if page_size > 0 else 0
  47. return PageResponse(
  48. code=0,
  49. data=data,
  50. pagination={"page": page, "page_size": page_size, "total": total, "total_pages": total_pages},
  51. ).model_dump()