mirror of
https://github.com/insistence/RuoYi-Vue3-FastAPI.git
synced 2026-09-22 13:05:19 +00:00
106 lines
3.1 KiB
Python
106 lines
3.1 KiB
Python
from datetime import datetime
|
|
from typing import Any, Generic, Optional, TypeVar, Union
|
|
|
|
from pydantic import BaseModel, ConfigDict, Field, create_model
|
|
from pydantic.alias_generators import to_camel
|
|
from typing_extensions import Self
|
|
|
|
from common.constant import HttpStatusConstant
|
|
|
|
T = TypeVar('T')
|
|
|
|
|
|
class CrudResponseModel(BaseModel):
|
|
"""
|
|
操作响应模型
|
|
"""
|
|
|
|
is_success: bool = Field(description='操作是否成功')
|
|
message: str = Field(description='响应信息')
|
|
result: Optional[Any] = Field(default=None, description='响应结果')
|
|
|
|
|
|
class ResponseBaseModel(BaseModel):
|
|
"""
|
|
响应模型
|
|
"""
|
|
|
|
code: int = Field(default=HttpStatusConstant.SUCCESS, description='响应码')
|
|
msg: str = Field(default='操作成功', description='响应信息')
|
|
success: bool = Field(default=True, description='响应是否成功')
|
|
time: datetime = Field(default_factory=datetime.now, description='响应时间')
|
|
|
|
|
|
class DynamicResponseModel(ResponseBaseModel, Generic[T]):
|
|
"""
|
|
动态响应模型
|
|
"""
|
|
|
|
model_config = ConfigDict(alias_generator=to_camel)
|
|
|
|
def __class_getitem__(cls, item: Any) -> Union[Any, Self]:
|
|
"""
|
|
当使用 DynamicResponseModel[Item] 语法时，动态创建一个包含所有字段的新模型
|
|
"""
|
|
# 检查是否已经为该类型创建了模型
|
|
if not hasattr(cls, '_cached_models'):
|
|
cls._cached_models = {}
|
|
|
|
if item in cls._cached_models:
|
|
return cls._cached_models[item]
|
|
|
|
# 检查item是否为Pydantic模型
|
|
if not hasattr(item, 'model_fields'):
|
|
raise TypeError(f'{item} 不是一个Pydantic模型，请使用Pydantic模型作为泛型参数')
|
|
|
|
# 获取ResponseBaseModel的字段
|
|
base_fields = {}
|
|
for field_name, field in cls.model_fields.items():
|
|
base_fields[field_name] = (field.annotation, field)
|
|
|
|
# 获取泛型类型的字段
|
|
item_fields = {}
|
|
for field_name, field in item.model_fields.items():
|
|
item_fields[field_name] = (field.annotation, field)
|
|
|
|
# 合并所有字段
|
|
all_fields = {**base_fields, **item_fields}
|
|
|
|
# 动态创建新模型
|
|
new_model = create_model(
|
|
f'DynamicResponseModel[{item.__name__}]', __base__=cls, __config__=cls.model_config, **all_fields
|
|
)
|
|
|
|
# 缓存模型
|
|
cls._cached_models[item] = new_model
|
|
|
|
return new_model
|
|
|
|
|
|
class PageModel(BaseModel, Generic[T]):
|
|
"""
|
|
分页模型
|
|
"""
|
|
|
|
model_config = ConfigDict(alias_generator=to_camel)
|
|
|
|
rows: list[T] = Field(description='记录列表')
|
|
page_num: int = Field(description='当前页码')
|
|
page_size: int = Field(description='每页记录数')
|
|
total: int = Field(description='总记录数')
|
|
has_next: bool = Field(description='是否有下一页')
|
|
|
|
|
|
class PageResponseModel(PageModel, ResponseBaseModel, Generic[T]):
|
|
"""
|
|
分页响应模型
|
|
"""
|
|
|
|
|
|
class DataResponseModel(ResponseBaseModel, Generic[T]):
|
|
"""
|
|
数据响应模型
|
|
"""
|
|
|
|
data: T = Field(description='响应数据')
|