from datetime import datetime from typing import Annotated, Generic, Literal, TypeVar from pydantic import ( AwareDatetime, BaseModel, ConfigDict, Field, HttpUrl, field_validator, model_validator, ) Title = Annotated[str, Field(min_length=1, max_length=250)] Category = Annotated[str, Field(min_length=1, max_length=80)] Body = Annotated[str, Field(min_length=1, max_length=100_000)] Notes = Annotated[str, Field(max_length=10_000)] Status = Literal["pending", "verified", "rejected"] PositiveId = Annotated[int, Field(gt=0)] class Input(BaseModel): model_config = ConfigDict(extra="forbid", str_strip_whitespace=True) @field_validator("source_url", mode="before", check_fields=False) @classmethod def empty_url_is_absent(cls, value): return None if isinstance(value, str) and not value.strip() else value class Patch(Input): expected_version: PositiveId @model_validator(mode="after") def require_changes(self): if not self.model_fields_set - {"expected_version"}: raise ValueError("至少提供一个需要更新的字段") return self @field_validator("*", mode="before") @classmethod def disallow_required_null(cls, value, info): if ( info.field_name in {"name", "title", "category", "content", "notes", "expected_version"} and value is None ): raise ValueError("该字段不能为 null") return value class Read(BaseModel): model_config = ConfigDict(from_attributes=True) id: int version: int created_at: datetime updated_at: datetime is_deleted: bool class SourceCreate(Input): name: Title source_url: HttpUrl | None = Field(default=None, max_length=1000) notes: Notes = "" class SourceUpdate(Patch): name: Title | None = None source_url: HttpUrl | None = Field(default=None, max_length=1000) notes: Notes | None = None class SourceRead(Read): name: str source_url: str | None notes: str class DocumentCreate(Input): source_id: PositiveId title: Title category: Category = "未分类" content: Body collected_at: AwareDatetime notes: Notes = "" class DocumentUpdate(Patch): title: Title | None = None category: Category | None = None content: Body | None = None collected_at: AwareDatetime | None = None notes: Notes | None = None @field_validator("collected_at", mode="before") @classmethod def collection_time_is_required(cls, value): if value is None: raise ValueError("文档采集时间不能为 null") return value class DocumentRead(Read): source_id: int title: str category: str content: str collected_at: datetime notes: str T = TypeVar("T") class Page(BaseModel, Generic[T]): items: list[T] total: int page: int page_size: int class AuditRead(BaseModel): model_config = ConfigDict(from_attributes=True) id: int entity_type: str entity_id: int action: str actor: str | None notes: str snapshot: dict created_at: datetime