strepsiades
141 lines
4.0 KiB
Python
141 lines
4.0 KiB
Python
from pydantic import BaseModel, Field, ValidationError, model_validator
|
||
from typing import Any, Dict
|
||
from typing import Any, Optional, TypeVar, Generic, List
|
||
from pydantic import BaseModel, Field, model_serializer
|
||
|
||
class BaseResponseWithID(BaseModel):
|
||
"""
|
||
Mixin: 统一处理输入中的 id/_id -> id
|
||
约束：
|
||
- 若 id 与 _id 同时出现且值不同 -> 报错
|
||
- 若仅 _id 出现 -> 转成 id
|
||
- 若两者都缺失 -> 报错
|
||
- 最终保证存在 id: str
|
||
"""
|
||
id: str = Field(..., description="资源唯一标识ID")
|
||
|
||
@model_validator(mode="before")
|
||
@classmethod
|
||
def normalize_id(cls, data: Any) -> Any:
|
||
# 仅处理 dict-like 输入；其他情况交给 Pydantic 默认流程
|
||
if not isinstance(data, dict):
|
||
return data
|
||
|
||
has_id = "id" in data
|
||
has__id = "_id" in data
|
||
|
||
if has_id and has__id:
|
||
if str(data["id"]) != str(data["_id"]):
|
||
raise ValidationError([
|
||
{
|
||
"type": "value_error",
|
||
"loc": ("id",),
|
||
"msg": "id and _id both provided but not equal",
|
||
"input": data.get("id"),
|
||
}
|
||
], cls)
|
||
# 相同则优先保留 id，去掉 _id
|
||
data = {**data}
|
||
data.pop("_id", None)
|
||
return data
|
||
|
||
if has__id and not has_id:
|
||
# 将 _id 统一映射为 id
|
||
data = {**data}
|
||
data["id"] = str(data["_id"])
|
||
data.pop("_id", None)
|
||
return data
|
||
|
||
if not has_id and not has__id:
|
||
raise ValidationError([
|
||
{
|
||
"type": "value_error.missing",
|
||
"loc": ("id",),
|
||
"msg": "id or _id is required",
|
||
"input": data,
|
||
}
|
||
], cls)
|
||
|
||
# 默认返回（已有 id）
|
||
return data
|
||
|
||
|
||
def _get_plan_node_date(node: Any) -> Any:
|
||
if isinstance(node, dict):
|
||
return node.get("date")
|
||
return getattr(node, "date", None)
|
||
|
||
|
||
def trim_plan_nodes_by_boundary_dates(
|
||
plan_nodes: Optional[List[Any]],
|
||
start_date: Optional[Any] = None,
|
||
end_date: Optional[Any] = None,
|
||
) -> List[Any]:
|
||
if not plan_nodes:
|
||
return []
|
||
|
||
cleaned_nodes = list(plan_nodes)
|
||
|
||
while (
|
||
start_date is not None
|
||
and cleaned_nodes
|
||
and str(_get_plan_node_date(cleaned_nodes[0])) == str(start_date)
|
||
):
|
||
cleaned_nodes.pop(0)
|
||
|
||
while (
|
||
end_date is not None
|
||
and cleaned_nodes
|
||
and str(_get_plan_node_date(cleaned_nodes[-1])) == str(end_date)
|
||
):
|
||
cleaned_nodes.pop()
|
||
|
||
return cleaned_nodes
|
||
|
||
class RequestApprovalMixin(BaseModel):
|
||
"""
|
||
请求审批相关的基础模型
|
||
"""
|
||
pass
|
||
|
||
T = TypeVar("T")
|
||
|
||
|
||
# def _build_full_response_dict(code: int, message: Optional[str], data: Any) -> dict:
|
||
# """构建与 __orjson__ 完全一致的响应字典"""
|
||
# # 处理 data 的序列化（与 __orjson__ 逻辑同步）
|
||
# if data is None:
|
||
# clean_data = {}
|
||
# elif isinstance(data, BaseModel):
|
||
# clean_data = data.model_dump()
|
||
# else:
|
||
# clean_data = data
|
||
# return {
|
||
# "code": code,
|
||
# "message": message,
|
||
# "data": clean_data
|
||
# }
|
||
|
||
class Success(BaseModel, Generic[T]):
|
||
code: int = 200
|
||
message: Optional[str] = "OK"
|
||
data: Optional[T] = None
|
||
|
||
# def __init__(self, **kwargs):
|
||
# super().__init__(**kwargs)
|
||
# # 自动捕获完整响应结构到 push_log.result
|
||
# push_log = current_push_log.get()
|
||
# if push_log is not None:
|
||
# full_response = _build_full_response_dict(self.code, self.message, self.data)
|
||
# push_log.result = full_response
|
||
#
|
||
class Fail(BaseModel):
|
||
code: int = 400
|
||
message: Optional[str] = None
|
||
data: Optional[Any] = Field(default_factory=dict)
|
||
|
||
class PaginatedList(BaseModel, Generic[T]):
|
||
list: List[T] = Field(default_factory=list)
|
||
total: int = 0
|
||
page: int = 1
|
||
page_size: int = 20 |