第 14 章文章数据处理综合实战

前面已经分别学过建模、校验、序列化、配置和数据场景拆分。最后一章把这些能力串起来，模拟一次轻量的文章数据处理：

准备应用配置
校验分页参数
校验文章创建数据
生成文章详情数据
导出字典和 JSON
整理模型错误和普通规则错误

输入正确时，流程会一路走到导出；遇到错误时，则整理出统一的结果。

用同一个处理入口验证四种输入：

场景	预期结果
分页与文章数据都正确	得到分页字典、文章详情字典和 JSON
`page_size=200`，配置上限为 `50`	返回 `page_size` 的规则错误
文章标题、内容、作者等字段不合格	返回带字段路径的校验错误
文章标记为已发布，但摘要为空	返回发布规则的模型级错误

这一章仍然保持轻量：

不连接数据库
不启动 Web 服务
不引入 ORM
不做认证、权限、日志系统

整章围绕一条完整的数据处理主线展开。

本章学习目标

最后用一个完整流程把各章串起来：组合 AppConfig、PageQuery、ArticleCreate 和 ArticleDetail，把输入校验与处理后的数据分层。你会分别整理 ValidationError 和普通 ValueError，并输出最终的字典与 JSON 结果。

场景与处理流程

先看整体流程：

文章数据处理综合实战端到端流水线架构图

主线会用到这些对象和函数：

text

AppConfig
PageQuery
UserInfo
ArticleCreate
ArticleDetail
validate_page_size()
create_article_detail()
format_validation_errors()
format_common_error()
process_article()
run_case()
display_errors()

process_article() 是统一入口；run_case() 用它跑四组输入；display_errors() 则负责筛出对外展示的错误字段。

每个示例都可以复制到 .code/main.py 运行；进入下一步时用新代码覆盖它。

固定配置与分页校验

固定配置与分页模型

先把已经学过的几步连起来看：

text

输入数据
-> 校验
-> 处理
-> 详情表达
-> 导出
-> 错误整理

固定配置参与分页规则

本章的配置值预先准备好，不从 .env 读取，所以用普通的 BaseModel 表示。上一章的 BaseSettings 才负责读取外部配置源。这里把这个固定配置模型叫作 AppConfig：

python

class AppConfig(BaseModel):
    app_name: str = "Article App"
    max_page_size: int = 100

接下来用它限制：

text

page_size 不能超过 max_page_size

综合处理流程

文章数据处理综合实战端到端流水线架构图

先准备配置和分页参数。把下面的完整代码保存为 .code/main.py：

from pydantic import BaseModel, Field, field_validator


class AppConfig(BaseModel):
    app_name: str = "Article App"
    max_page_size: int = Field(default=100, ge=1)


class PageQuery(BaseModel):
    page: int = Field(default=1, ge=1)
    page_size: int = Field(default=10, ge=1)
    keyword: str | None = Field(default=None)

    @field_validator("keyword")
    @classmethod
    def clean_keyword(cls, value: str | None) -> str | None:
        if value is None:
            return value
        value = value.strip()
        return value or None


def validate_page_size(query: PageQuery, config: AppConfig) -> None:
    if query.page_size > config.max_page_size:
        raise ValueError(f"page_size 不能超过 {config.max_page_size}")


config = AppConfig(max_page_size=50)

query_data = {
    "page": "1",
    "page_size": "20",
    "keyword": "  Pydantic  ",
}

query = PageQuery.model_validate(query_data)
validate_page_size(query, config)

print(config.app_name)
print(query.model_dump(exclude_none=True))

运行：

bash

uv run python main.py

运行结果：

text

Article App
{'page': 1, 'page_size': 20, 'keyword': 'Pydantic'}

分页运行结果与规则

固定配置 AppConfig

这里用它保存：

应用名称
最大分页大小

分页输入 PageQuery

字段包括：

page
page_size
keyword

搜索词清理

把：

python

"  Pydantic  "

变成：

text

Pydantic

分页上限校验

模型负责基础校验，AppConfig 提供最大值；两步都通过后，分页参数才能进入后续流程。

配置与分页流程

四组测试用例分支验证与拦截流转矩阵

文章创建校验

创建模型与运行结果

分页参数和文章创建数据都来自外部输入，先交给模型校验：

例如：

python

query = PageQuery.model_validate(query_data)
article_create = ArticleCreate.model_validate(article_data)

接着校验文章创建数据。将完整代码保存为 .code/main.py：

from pydantic import BaseModel, Field, field_validator


class UserInfo(BaseModel):
    username: str = Field(min_length=2, max_length=20)
    email: str

    @field_validator("username", mode="before")
    @classmethod
    def validate_username(cls, value: object) -> object:
        if not isinstance(value, str):
            return value
        value = value.strip()
        if " " in value:
            raise ValueError("用户名不能包含空格")
        return value

    @field_validator("email")
    @classmethod
    def validate_email(cls, value: str) -> str:
        name, separator, domain = value.partition("@")
        if not name or not separator or not domain or "@" in domain:
            raise ValueError("邮箱必须包含用户名和域名")
        return value


class ArticleCreate(BaseModel):
    title: str = Field(min_length=2, max_length=50)
    content: str = Field(min_length=10)
    tags: list[str] = Field(default_factory=list, max_length=5)
    summary: str | None = Field(default=None, max_length=100)
    author: UserInfo

    @field_validator("title")
    @classmethod
    def validate_title(cls, value: str) -> str:
        blocked_words = ["广告", "推广"]
        for word in blocked_words:
            if word in value:
                raise ValueError("文章标题不能包含敏感词")
        return value


article_data = {
    "title": "Pydantic 综合实战",
    "content": "这是一篇用于演示 Pydantic 综合实战流程的文章内容",
    "tags": ["Python", "Pydantic"],
    "summary": "把数据校验和序列化流程串起来",
    "author": {
        "username": "xiaoming",
        "email": "xiaoming@example.com",
    },
}

article_create = ArticleCreate.model_validate(article_data)

print(article_create.title)
print(article_create.author.username)
print(article_create.model_dump())

运行：

bash

uv run python main.py

输出依次是标题、作者用户名和包含嵌套 author 的完整字典。这里截取前两行：

text

Pydantic 综合实战
xiaoming

作者与字段规则

作者模型 UserInfo

作者有自己的结构和校验规则，所以用 UserInfo 表达，不直接写成裸 dict。

用户名与邮箱校验

用户名和邮箱各自的规则都属于单字段校验。

创建模型 ArticleCreate

创建时不要求用户提交：

id
views
created_at

嵌套作者信息

嵌套作者数据也会按 UserInfo 校验。后续生成详情时，基于已校验的创建模型继续处理，不直接拼接原始输入。

创建数据流程

四组测试用例分支验证与拦截流转矩阵

文章详情生成与导出

详情模型与处理函数

创建数据是用户提交的内容；详情数据是程序补充字段后得到的完整记录。因此继续使用两个模型：

ArticleCreate
ArticleDetail

详情导出

处理并校验完成后，可以按需要导出成字典或 JSON：

python

article_detail.model_dump()
article_detail.model_dump_json(indent=2)

下面来生成“处理后数据”。将完整代码保存为 .code/main.py：

from datetime import datetime

from pydantic import BaseModel, Field, field_serializer, field_validator, model_validator


class UserInfo(BaseModel):
    username: str = Field(min_length=2, max_length=20)
    email: str

    @field_validator("username", mode="before")
    @classmethod
    def validate_username(cls, value: object) -> object:
        if not isinstance(value, str):
            return value
        value = value.strip()
        if " " in value:
            raise ValueError("用户名不能包含空格")
        return value

    @field_validator("email")
    @classmethod
    def validate_email(cls, value: str) -> str:
        name, separator, domain = value.partition("@")
        if not name or not separator or not domain or "@" in domain:
            raise ValueError("邮箱必须包含用户名和域名")
        return value


class ArticleCreate(BaseModel):
    title: str = Field(min_length=2, max_length=50)
    content: str = Field(min_length=10)
    tags: list[str] = Field(default_factory=list, max_length=5)
    summary: str | None = Field(default=None, max_length=100)
    author: UserInfo

    @field_validator("title")
    @classmethod
    def validate_title(cls, value: str) -> str:
        blocked_words = ["广告", "推广"]
        for word in blocked_words:
            if word in value:
                raise ValueError("文章标题不能包含敏感词")
        return value


class ArticleDetail(ArticleCreate):
    id: int
    views: int = Field(default=0, ge=0)
    is_published: bool = Field(default=False)
    created_at: datetime

    @model_validator(mode="after")
    def validate_publish_rules(self):
        if self.is_published and not self.summary:
            raise ValueError("文章发布时必须填写摘要")
        return self

    @field_serializer("created_at")
    def serialize_created_at(self, value: datetime) -> str:
        return value.strftime("%Y-%m-%d %H:%M")


def create_article_detail(article_create: ArticleCreate) -> ArticleDetail:
    detail_data = {
        **article_create.model_dump(),
        "id": 1,
        "views": 0,
        "is_published": True,
        "created_at": datetime.now(),
    }
    return ArticleDetail.model_validate(detail_data)


article_data = {
    "title": "Pydantic 综合实战",
    "content": "这是一篇用于演示 Pydantic 综合实战流程的文章内容",
    "tags": ["Python", "Pydantic"],
    "summary": "把数据校验和序列化流程串起来",
    "author": {
        "username": "xiaoming",
        "email": "xiaoming@example.com",
    },
}

article_create = ArticleCreate.model_validate(article_data)
article_detail = create_article_detail(article_create)

print(article_detail.model_dump())
print(article_detail.model_dump_json(indent=2))

运行：

bash

uv run python main.py

python

from datetime import datetime

from pydantic import BaseModel, Field, field_serializer, field_validator, model_validator


class UserInfo(BaseModel):
    username: str = Field(min_length=2, max_length=20)
    email: str

    @field_validator("username", mode="before")
    @classmethod
    def validate_username(cls, value: object) -> object:
        if not isinstance(value, str):
            return value
        value = value.strip()
        if " " in value:
            raise ValueError("用户名不能包含空格")
        return value

    @field_validator("email")
    @classmethod
    def validate_email(cls, value: str) -> str:
        name, separator, domain = value.partition("@")
        if not name or not separator or not domain or "@" in domain:
            raise ValueError("邮箱必须包含用户名和域名")
        return value


class ArticleCreate(BaseModel):
    title: str = Field(min_length=2, max_length=50)
    content: str = Field(min_length=10)
    tags: list[str] = Field(default_factory=list, max_length=5)
    summary: str | None = Field(default=None, max_length=100)
    author: UserInfo

    @field_validator("title")
    @classmethod
    def validate_title(cls, value: str) -> str:
        blocked_words = ["广告", "推广"]
        for word in blocked_words:
            if word in value:
                raise ValueError("文章标题不能包含敏感词")
        return value


class ArticleDetail(ArticleCreate):
    id: int
    views: int = Field(default=0, ge=0)
    is_published: bool = Field(default=False)
    created_at: datetime

    @model_validator(mode="after")
    def validate_publish_rules(self):
        if self.is_published and not self.summary:
            raise ValueError("文章发布时必须填写摘要")
        return self

    @field_serializer("created_at")
    def serialize_created_at(self, value: datetime) -> str:
        return value.strftime("%Y-%m-%d %H:%M")


def create_article_detail(article_create: ArticleCreate) -> ArticleDetail:
    detail_data = {
        **article_create.model_dump(),
        "id": 1,
        "views": 0,
        "is_published": True,
        "created_at": datetime.now(),
    }
    return ArticleDetail.model_validate(detail_data)


article_data = {
    "title": "Pydantic 综合实战",
    "content": "这是一篇用于演示 Pydantic 综合实战流程的文章内容",
    "tags": ["Python", "Pydantic"],
    "summary": "把数据校验和序列化流程串起来",
    "author": {
        "username": "xiaoming",
        "email": "xiaoming@example.com",
    },
}

article_create = ArticleCreate.model_validate(article_data)
article_detail = create_article_detail(article_create)

print(article_detail.model_dump())
print(article_detail.model_dump_json(indent=2))

字典和 JSON 中的 created_at 都会按 %Y-%m-%d %H:%M 格式输出；具体时间随运行时刻变化，模型内部仍保存 datetime 对象。

导出结果与发布规则

继承创建字段

继承 ArticleCreate，就能复用创建阶段定义的字段。

详情生成函数

函数负责：

text

接收 ArticleCreate
-> 补充详情字段
-> 返回 ArticleDetail

发布规则

例如：

text

已发布文章必须有摘要

创建时间序列化

created_at 在模型中仍是 datetime，导出时才会变成字符串。

补充详情字段后还会重新校验，这样得到的是经过检查的 ArticleDetail，不只是一个拼接好的字典。

详情生成与导出流程

四组测试用例分支验证与拦截流转矩阵

模型与规则错误整理

两类错误的统一结构

如果只打印异常：

python

print(e)

大致能看出问题，但后续展示和处理不太方便。本章会把错误整理成统一结构，例如：

python

{
    "field": "author.email",
    "message": "Value error, 邮箱必须包含用户名和域名",
    "input": "not-email",
}

流程里会遇到两类错误：

Pydantic 模型校验错误
普通规则函数抛出的 ValueError

因此分别用两个函数整理。将下面的完整代码保存为 .code/main.py：

from datetime import datetime

from pydantic import BaseModel, Field, ValidationError, field_serializer, field_validator, model_validator


class UserInfo(BaseModel):
    username: str = Field(min_length=2, max_length=20)
    email: str

    @field_validator("username", mode="before")
    @classmethod
    def validate_username(cls, value: object) -> object:
        if not isinstance(value, str):
            return value
        value = value.strip()
        if " " in value:
            raise ValueError("用户名不能包含空格")
        return value

    @field_validator("email")
    @classmethod
    def validate_email(cls, value: str) -> str:
        name, separator, domain = value.partition("@")
        if not name or not separator or not domain or "@" in domain:
            raise ValueError("邮箱必须包含用户名和域名")
        return value


class ArticleCreate(BaseModel):
    title: str = Field(min_length=2, max_length=50)
    content: str = Field(min_length=10)
    tags: list[str] = Field(default_factory=list, max_length=5)
    summary: str | None = Field(default=None, max_length=100)
    author: UserInfo

    @field_validator("title")
    @classmethod
    def validate_title(cls, value: str) -> str:
        blocked_words = ["广告", "推广"]
        for word in blocked_words:
            if word in value:
                raise ValueError("文章标题不能包含敏感词")
        return value


class ArticleDetail(ArticleCreate):
    id: int
    views: int = Field(default=0, ge=0)
    is_published: bool = Field(default=False)
    created_at: datetime

    @model_validator(mode="after")
    def validate_publish_rules(self):
        if self.is_published and not self.summary:
            raise ValueError("文章发布时必须填写摘要")
        return self

    @field_serializer("created_at")
    def serialize_created_at(self, value: datetime) -> str:
        return value.strftime("%Y-%m-%d %H:%M")


def create_article_detail(article_create: ArticleCreate) -> ArticleDetail:
    detail_data = {
        **article_create.model_dump(),
        "id": 1,
        "views": 0,
        "is_published": True,
        "created_at": datetime.now(),
    }
    return ArticleDetail.model_validate(detail_data)


def format_validation_errors(error: ValidationError) -> list[dict]:
    result = []

    for item in error.errors():
        result.append(
            {
                "field": ".".join(str(loc) for loc in item["loc"]) or "模型",
                "message": item["msg"],
                "input": item.get("input"),
            }
        )

    return result


def format_common_error(
    field: str, error: ValueError, input_value: object | None = None
) -> dict:
    return {
        "field": field,
        "message": str(error),
        "input": input_value,
    }


bad_article_data = {
    "title": "广",
    "content": "太短",
    "tags": ["Python", "Pydantic", "实战", "校验", "错误", "超出"],
    "summary": None,
    "author": {
        "username": "xiao ming",
        "email": "not-email",
    },
}

try:
    article_create = ArticleCreate.model_validate(bad_article_data)
    article_detail = create_article_detail(article_create)
    print(article_detail.model_dump())
except ValidationError as error:
    print(format_validation_errors(error))

bad_detail_data = {
    "title": "Pydantic 综合实战",
    "content": "这是一篇用于演示发布规则错误的文章内容",
    "tags": ["Python", "Pydantic"],
    "summary": None,
    "author": {"username": "xiaoming", "email": "xiaoming@example.com"},
    "id": 1,
    "is_published": True,
    "created_at": datetime.now(),
}

try:
    ArticleDetail.model_validate(bad_detail_data)
except ValidationError as error:
    for item in format_validation_errors(error):
        print({"field": item["field"], "message": item["message"]})


def validate_page_size(page_size: int, max_page_size: int) -> None:
    if page_size > max_page_size:
        raise ValueError(f"page_size 不能超过 {max_page_size}")


try:
    page_size = 200
    validate_page_size(page_size, max_page_size=100)
except ValueError as error:
    print(format_common_error("page_size", error, page_size))

运行：

bash

uv run python main.py

python

from pydantic import ValidationError


def format_validation_errors(error: ValidationError) -> list[dict]:
    result = []

    for item in error.errors():
        result.append({
            "field": ".".join(str(loc) for loc in item["loc"]) or "模型",
            "message": item["msg"],
            "input": item.get("input"),
        })

    return result


def format_common_error(
        field: str, error: ValueError, input_value: object | None = None
) -> dict:
    return {
        "field": field,
        "message": str(error),
        "input": input_value,
    }

普通规则错误也保留这三个键。分页上限错误可以这样整理：

python

format_common_error("page_size", ValueError("page_size 不能超过 100"), 200)

结果中 input 是 200，本地排错时可以用它定位问题。给界面或调用方展示时，只取 field 和 message，不带原始输入。

再试一组有问题的文章数据：

python

bad_article_data = {
    "title": "广",
    "content": "太短",
    "tags": ["Python", "Pydantic", "实战", "校验", "错误", "超出"],
    "summary": None,
    "author": {
        "username": "xiao ming",
        "email": "not-email",
    },
}

try:
    article_create = ArticleCreate.model_validate(bad_article_data)
    article_detail = create_article_detail(article_create)
    print(article_detail.model_dump())
except ValidationError as e:
    print(format_validation_errors(e))

附件还会触发分页超限和发布规则错误。发布规则属于模型级校验，loc 是空元组；整理函数会将它标为“模型”：

text

{'field': '模型', 'message': 'Value error, 文章发布时必须填写摘要'}

对外只展示字段名和说明；原始输入仍在内部整理结果中，供本地排错。

错误位置与展示边界

模型校验错误

从 e.errors() 中提取：

字段路径
错误说明
原始输入

错误字段路径

例如：

text

author.email

模型级错误没有字段路径，用“模型”标记。

普通规则错误

分页大小超过配置上限时，就走普通错误函数。传入输入值后，两类错误都会整理成含 field、message 和 input 的结构。

统一结构

展示时按场景选择字段，内部仍沿用统一结构。

错误整理流程

统一异常收敛与格式化流程图

文章处理综合案例

最后把这条主线放进统一入口，跑完四组输入：

统一处理入口

将下面的完整代码保存为 .code/main.py：

from datetime import datetime

from pydantic import (
    BaseModel,
    Field,
    ValidationError,
    field_serializer,
    field_validator,
    model_validator,
)


class AppConfig(BaseModel):
    app_name: str = "Article App"
    max_page_size: int = Field(default=100, ge=1)


class PageQuery(BaseModel):
    page: int = Field(default=1, ge=1)
    page_size: int = Field(default=10, ge=1)
    keyword: str | None = Field(default=None)

    @field_validator("keyword")
    @classmethod
    def clean_keyword(cls, value: str | None) -> str | None:
        if value is None:
            return value
        value = value.strip()
        return value or None


class UserInfo(BaseModel):
    username: str = Field(min_length=2, max_length=20)
    email: str

    @field_validator("username", mode="before")
    @classmethod
    def validate_username(cls, value: object) -> object:
        if not isinstance(value, str):
            return value
        value = value.strip()
        if " " in value:
            raise ValueError("用户名不能包含空格")
        return value

    @field_validator("email")
    @classmethod
    def validate_email(cls, value: str) -> str:
        name, separator, domain = value.partition("@")
        if not name or not separator or not domain or "@" in domain:
            raise ValueError("邮箱必须包含用户名和域名")
        return value


class ArticleCreate(BaseModel):
    title: str = Field(min_length=2, max_length=50)
    content: str = Field(min_length=10)
    tags: list[str] = Field(default_factory=list, max_length=5)
    summary: str | None = Field(default=None, max_length=100)
    author: UserInfo

    @field_validator("title")
    @classmethod
    def validate_title(cls, value: str) -> str:
        blocked_words = ["广告", "推广"]
        for word in blocked_words:
            if word in value:
                raise ValueError("文章标题不能包含敏感词")
        return value


class ArticleDetail(ArticleCreate):
    id: int
    views: int = Field(default=0, ge=0)
    is_published: bool = Field(default=False)
    created_at: datetime

    @model_validator(mode="after")
    def validate_publish_rules(self):
        if self.is_published and not self.summary:
            raise ValueError("文章发布时必须填写摘要")
        return self

    @field_serializer("created_at")
    def serialize_created_at(self, value: datetime) -> str:
        return value.strftime("%Y-%m-%d %H:%M")


def validate_page_size(query: PageQuery, config: AppConfig) -> None:
    if query.page_size > config.max_page_size:
        raise ValueError(f"page_size 不能超过 {config.max_page_size}")


def create_article_detail(article_create: ArticleCreate) -> ArticleDetail:
    detail_data = {
        **article_create.model_dump(),
        "id": 1,
        "views": 0,
        "is_published": True,
        "created_at": datetime.now(),
    }
    return ArticleDetail.model_validate(detail_data)


def format_validation_errors(error: ValidationError) -> list[dict]:
    result = []
    for item in error.errors():
        result.append(
            {
                "field": ".".join(str(loc) for loc in item["loc"]) or "模型",
                "message": item["msg"],
                "input": item.get("input"),
            }
        )
    return result


def format_common_error(
    field: str, error: ValueError, input_value: object | None = None
) -> dict:
    return {
        "field": field,
        "message": str(error),
        "input": input_value,
    }


def display_errors(items: list[dict]) -> list[dict]:
    return [
        {"field": item["field"], "message": item["message"]}
        for item in items
    ]


def process_article(
    query_data: dict, article_data: dict, config: AppConfig
) -> dict:
    query = PageQuery.model_validate(query_data)
    validate_page_size(query, config)

    article_create = ArticleCreate.model_validate(article_data)
    article_detail = create_article_detail(article_create)

    return {
        "app_name": config.app_name,
        "query": query.model_dump(exclude_none=True),
        "article": article_detail.model_dump(),
        "article_json": article_detail.model_dump_json(indent=2),
    }


def run_case(
    name: str, query_data: dict, article_data: dict, config: AppConfig
) -> None:
    print(f"【{name}】")
    try:
        result = process_article(query_data, article_data, config)
    except ValidationError as error:
        print(display_errors(format_validation_errors(error)))
    except ValueError as error:
        item = format_common_error("page_size", error, query_data.get("page_size"))
        print(display_errors([item]))
    else:
        print(result["app_name"])
        print(result["query"])
        print(result["article"])
        print(result["article_json"])


config = AppConfig(app_name="Article App", max_page_size=50)

query_data = {
    "page": "1",
    "page_size": "20",
    "keyword": "  Pydantic  ",
}

article_data = {
    "title": "Pydantic 综合实战",
    "content": "这是一篇用于演示 Pydantic 综合实战流程的文章内容",
    "tags": ["Python", "Pydantic"],
    "summary": "把数据校验和序列化流程串起来",
    "author": {
        "username": "xiaoming",
        "email": "xiaoming@example.com",
    },
}

bad_query_data = {
    "page": 1,
    "page_size": 200,
    "keyword": "Pydantic",
}

bad_article_data = {
    "title": "广",
    "content": "太短",
    "tags": ["Python", "Pydantic", "实战", "校验", "错误", "超出"],
    "summary": None,
    "author": {
        "username": "xiao ming",
        "email": "not-email",
    },
}

bad_detail_article_data = {**article_data, "summary": None}

run_case("正常输入", query_data, article_data, config)
run_case("分页超限", bad_query_data, article_data, config)
run_case("创建数据错误", query_data, bad_article_data, config)
run_case("发布规则错误", query_data, bad_detail_article_data, config)

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211

运行：

bash

uv run python main.py

python

from datetime import datetime

from pydantic import (
    BaseModel,
    Field,
    ValidationError,
    field_serializer,
    field_validator,
    model_validator,
)


class AppConfig(BaseModel):
    app_name: str = "Article App"
    max_page_size: int = Field(default=100, ge=1)


class PageQuery(BaseModel):
    page: int = Field(default=1, ge=1)
    page_size: int = Field(default=10, ge=1)
    keyword: str | None = Field(default=None)

    @field_validator("keyword")
    @classmethod
    def clean_keyword(cls, value: str | None) -> str | None:
        if value is None:
            return value
        value = value.strip()
        return value or None


class UserInfo(BaseModel):
    username: str = Field(min_length=2, max_length=20)
    email: str

    @field_validator("username", mode="before")
    @classmethod
    def validate_username(cls, value: object) -> object:
        if not isinstance(value, str):
            return value
        value = value.strip()
        if " " in value:
            raise ValueError("用户名不能包含空格")
        return value

    @field_validator("email")
    @classmethod
    def validate_email(cls, value: str) -> str:
        name, separator, domain = value.partition("@")
        if not name or not separator or not domain or "@" in domain:
            raise ValueError("邮箱必须包含用户名和域名")
        return value


class ArticleCreate(BaseModel):
    title: str = Field(min_length=2, max_length=50)
    content: str = Field(min_length=10)
    tags: list[str] = Field(default_factory=list, max_length=5)
    summary: str | None = Field(default=None, max_length=100)
    author: UserInfo

    @field_validator("title")
    @classmethod
    def validate_title(cls, value: str) -> str:
        blocked_words = ["广告", "推广"]
        for word in blocked_words:
            if word in value:
                raise ValueError("文章标题不能包含敏感词")
        return value


class ArticleDetail(ArticleCreate):
    id: int
    views: int = Field(default=0, ge=0)
    is_published: bool = Field(default=False)
    created_at: datetime

    @model_validator(mode="after")
    def validate_publish_rules(self):
        if self.is_published and not self.summary:
            raise ValueError("文章发布时必须填写摘要")
        return self

    @field_serializer("created_at")
    def serialize_created_at(self, value: datetime) -> str:
        return value.strftime("%Y-%m-%d %H:%M")


def validate_page_size(query: PageQuery, config: AppConfig) -> None:
    if query.page_size > config.max_page_size:
        raise ValueError(f"page_size 不能超过 {config.max_page_size}")


def create_article_detail(article_create: ArticleCreate) -> ArticleDetail:
    detail_data = {
        **article_create.model_dump(),
        "id": 1,
        "views": 0,
        "is_published": True,
        "created_at": datetime.now(),
    }
    return ArticleDetail.model_validate(detail_data)


def format_validation_errors(error: ValidationError) -> list[dict]:
    result = []
    for item in error.errors():
        result.append(
            {
                "field": ".".join(str(loc) for loc in item["loc"]) or "模型",
                "message": item["msg"],
                "input": item.get("input"),
            }
        )
    return result


def format_common_error(
    field: str, error: ValueError, input_value: object | None = None
) -> dict:
    return {
        "field": field,
        "message": str(error),
        "input": input_value,
    }


def display_errors(items: list[dict]) -> list[dict]:
    return [
        {"field": item["field"], "message": item["message"]}
        for item in items
    ]


def process_article(
    query_data: dict, article_data: dict, config: AppConfig
) -> dict:
    query = PageQuery.model_validate(query_data)
    validate_page_size(query, config)

    article_create = ArticleCreate.model_validate(article_data)
    article_detail = create_article_detail(article_create)

    return {
        "app_name": config.app_name,
        "query": query.model_dump(exclude_none=True),
        "article": article_detail.model_dump(),
        "article_json": article_detail.model_dump_json(indent=2),
    }


def run_case(
    name: str, query_data: dict, article_data: dict, config: AppConfig
) -> None:
    print(f"【{name}】")
    try:
        result = process_article(query_data, article_data, config)
    except ValidationError as error:
        print(display_errors(format_validation_errors(error)))
    except ValueError as error:
        item = format_common_error("page_size", error, query_data.get("page_size"))
        print(display_errors([item]))
    else:
        print(result["app_name"])
        print(result["query"])
        print(result["article"])
        print(result["article_json"])


config = AppConfig(app_name="Article App", max_page_size=50)

query_data = {
    "page": "1",
    "page_size": "20",
    "keyword": "  Pydantic  ",
}

article_data = {
    "title": "Pydantic 综合实战",
    "content": "这是一篇用于演示 Pydantic 综合实战流程的文章内容",
    "tags": ["Python", "Pydantic"],
    "summary": "把数据校验和序列化流程串起来",
    "author": {
        "username": "xiaoming",
        "email": "xiaoming@example.com",
    },
}

bad_query_data = {
    "page": 1,
    "page_size": 200,
    "keyword": "Pydantic",
}

bad_article_data = {
    "title": "广",
    "content": "太短",
    "tags": ["Python", "Pydantic", "实战", "校验", "错误", "超出"],
    "summary": None,
    "author": {
        "username": "xiao ming",
        "email": "not-email",
    },
}

bad_detail_article_data = {**article_data, "summary": None}

run_case("正常输入", query_data, article_data, config)
run_case("分页超限", bad_query_data, article_data, config)
run_case("创建数据错误", query_data, bad_article_data, config)
run_case("发布规则错误", query_data, bad_detail_article_data, config)

四组输入与运行结果

四组输入都经过 process_article()：成功时得到分页字典、详情字典和 JSON；失败时，普通 ValueError 与模型 ValidationError 会先整理成相同结构，再由 display_errors() 只输出字段名和说明。

运行结果中的关键部分如下。创建时间取决于运行时刻，详情字典和 JSON 中均为格式化字符串。

text

【正常输入】
Article App
{'page': 1, 'page_size': 20, 'keyword': 'Pydantic'}
【分页超限】
[{'field': 'page_size', 'message': 'page_size 不能超过 50'}]
【发布规则错误】
[{'field': '模型', 'message': 'Value error, 文章发布时必须填写摘要'}]

“创建数据错误”还会列出 title、content、tags、author.username 和 author.email。内部整理结果保留 input 供本地排错；这段展示结果不带原始输入。

处理流程与代码结构

完整案例执行流程

文章数据处理综合实战端到端流水线架构图

代码结构回顾

可以把这一章的代码分成 4 层：

数据处理四层软件架构蓝图

这不是大型工程架构，只是把教学代码分层摆放，方便看清各部分的职责。

本章小结

这一章把配置、输入校验、业务处理、详情模型、导出和错误展示接成一条线。AppConfig 保存准备好的固定值，PageQuery、ArticleCreate、ArticleDetail 各管自己的数据边界；处理结果继续用模型表达，失败时交给错误整理函数。

第 14 章 文章数据处理综合实战 ​

本章学习目标 ​

场景与处理流程 ​

固定配置与分页校验 ​

固定配置与分页模型 ​

分页运行结果与规则 ​

文章创建校验 ​

创建模型与运行结果 ​

作者与字段规则 ​

文章详情生成与导出 ​

详情模型与处理函数 ​

导出结果与发布规则 ​

模型与规则错误整理 ​

两类错误的统一结构 ​

错误位置与展示边界 ​

文章处理综合案例 ​

统一处理入口 ​

四组输入与运行结果 ​

处理流程与代码结构 ​

本章小结 ​

第 14 章文章数据处理综合实战

本章学习目标

场景与处理流程

固定配置与分页校验

固定配置与分页模型

分页运行结果与规则

文章创建校验

创建模型与运行结果

作者与字段规则

文章详情生成与导出

详情模型与处理函数

导出结果与发布规则

模型与规则错误整理

两类错误的统一结构

错误位置与展示边界

文章处理综合案例

统一处理入口

四组输入与运行结果

处理流程与代码结构

本章小结