103 lines
2.3 KiB
Python
103 lines
2.3 KiB
Python
"""
|
|
基础 Schema 定义
|
|
|
|
定义通用的响应模式和基础配置。
|
|
"""
|
|
|
|
from datetime import datetime
|
|
from typing import Any, Generic, TypeVar
|
|
|
|
from pydantic import BaseModel, ConfigDict
|
|
|
|
DataT = TypeVar("DataT")
|
|
|
|
|
|
class BaseSchema(BaseModel):
|
|
"""基础 Schema 配置"""
|
|
|
|
model_config = ConfigDict(
|
|
from_attributes=True, # 支持从 ORM 对象创建
|
|
populate_by_name=True, # 支持字段别名
|
|
str_strip_whitespace=True, # 自动去除字符串首尾空白
|
|
)
|
|
|
|
|
|
class TimestampMixin(BaseModel):
|
|
"""时间戳混入"""
|
|
|
|
created_at: datetime
|
|
updated_at: datetime
|
|
|
|
|
|
class APIResponse(BaseModel, Generic[DataT]):
|
|
"""统一 API 响应格式"""
|
|
|
|
success: bool = True
|
|
message: str = "操作成功"
|
|
data: DataT | None = None
|
|
|
|
@classmethod
|
|
def ok(cls, data: DataT | None = None, message: str = "操作成功") -> "APIResponse[DataT]":
|
|
"""成功响应"""
|
|
return cls(success=True, message=message, data=data)
|
|
|
|
@classmethod
|
|
def error(cls, message: str = "操作失败", data: DataT | None = None) -> "APIResponse[DataT]":
|
|
"""错误响应"""
|
|
return cls(success=False, message=message, data=data)
|
|
|
|
|
|
class ErrorResponse(BaseModel):
|
|
"""错误响应"""
|
|
|
|
success: bool = False
|
|
message: str
|
|
code: str
|
|
details: dict[str, Any] = {}
|
|
|
|
|
|
class PaginationParams(BaseModel):
|
|
"""分页参数"""
|
|
|
|
page: int = 1
|
|
page_size: int = 20
|
|
|
|
@property
|
|
def offset(self) -> int:
|
|
"""计算偏移量"""
|
|
return (self.page - 1) * self.page_size
|
|
|
|
@property
|
|
def limit(self) -> int:
|
|
"""获取限制数量"""
|
|
return self.page_size
|
|
|
|
|
|
class PaginatedResponse(BaseModel, Generic[DataT]):
|
|
"""分页响应"""
|
|
|
|
items: list[DataT]
|
|
total: int
|
|
page: int
|
|
page_size: int
|
|
total_pages: int
|
|
|
|
@classmethod
|
|
def create(
|
|
cls,
|
|
items: list[DataT],
|
|
total: int,
|
|
page: int,
|
|
page_size: int,
|
|
) -> "PaginatedResponse[DataT]":
|
|
"""创建分页响应"""
|
|
total_pages = (total + page_size - 1) // page_size if page_size > 0 else 0
|
|
return cls(
|
|
items=items,
|
|
total=total,
|
|
page=page,
|
|
page_size=page_size,
|
|
total_pages=total_pages,
|
|
)
|
|
|