131 lines
3.1 KiB
Python
131 lines
3.1 KiB
Python
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
|