Litestar DTO 教程用 DTO 工厂构建灵活的数据传输层【免费下载链接】litestarLight, flexible and extensible ASGI framework | Built to scale项目地址: https://gitcode.com/GitHub_Trending/li/litestar本篇为 Litestar 官方 DTO 教程Data Transfer Object Tutorial的中文技术详解。教程面向已熟悉 Litestar 路由处理器route handler基础用法的开发者完整覆盖从返回 dataclass 的朴素模式到基于 DTO 的接收、校验、隐藏、重命名与分层复用的全过程。读完本篇你将掌握DataclassDTO、DTOConfig、DTOData三个核心工具的实际用法能独立用 DTO 工厂实现响应字段裁剪、请求数据校验、只读字段、PUT/PATCH 更新以及 Controller 分层 DTO 声明等实战能力。前置要求与阅读路径本教程假设你已经熟悉 Litestar 及路由处理器等基础概念。如果尚未接触建议先阅读仓库中的 TODO 应用基础教程Developing a basic TODO application后再回到本篇。教程的全部可运行示例代码位于仓库的docs/examples/data_transfer_objects/factory/tutorial/目录下每节内容都有对应的独立脚本核心 API 的参考文档可查阅 docs/reference/dto。1. 起步返回 dataclass 的朴素模式我们从一个最简单的应用开始。定义一个名为Person的 Pythondataclass数据模型包含name、age、email三个属性再定义一个路径为/person/{name:str}的 GET 路由处理器路径中的{name:str}表示名为name的字符串类型路径参数。最后创建Litestar应用实例并注册该路由处理器。完整代码见 initial_pattern.pyfrom __future__ import annotations from dataclasses import dataclass from litestar import Litestar, get from litestar.params import FromPath dataclass class Person: name: str age: int email: str get(/person/{name:str}, sync_to_threadFalse) def get_person(name: FromPath[str]) - Person: # Your logic to retrieve the person goes here # For demonstration purposes, a placeholder Person instance is returned return Person(namename, age30, emailfemail_of_{name}example.com) app Litestar(route_handlers[get_person])在 Litestar 中这种模式是开箱即用的从路由处理器返回 dataclass 实例是原生支持的行为。Litestar 会自动将该 dataclass 实例序列化为可通过网络传输的bytes默认输出为 JSON。将上面的脚本保存为app.py使用litestar run命令启动然后访问http://localhost:8000/person/peter浏览器中会看到类似下面的输出可以看到返回的 JSON 中包含了name、age、email全部字段。代码中的FromPath[str]用于声明路径参数并完成类型转换sync_to_threadFalse则让同步处理器直接在线程内执行避免同步函数被调度到线程池适用于本例这类纯计算场景。不过真实世界的应用很少这么简单。例如我们可能希望在用户创建之后限制对外暴露的信息——比如把用户的邮箱从响应中隐藏掉。这正是 Data Transfer Object数据传输对象要解决的问题。2. 第一个 DTO隐藏 email 字段我们在脚本中引入一个 DTO 类ReadDTO将其配置为排除Person.email字段并让路由处理器使用该 DTO 处理响应。完整代码见 simple_dto_exclude.pyfrom __future__ import annotations from dataclasses import dataclass from litestar import Litestar, get from litestar.dto import DataclassDTO, DTOConfig from litestar.params import FromPath dataclass class Person: name: str age: int email: str class ReadDTO(DataclassDTO[Person]): config DTOConfig(exclude{email}) get(/person/{name:str}, return_dtoReadDTO, sync_to_threadFalse) def get_person(name: FromPath[str]) - Person: # Your logic to retrieve the person goes here # For demonstration purposes, a placeholder Person instance is returned return Person(namename, age30, emailfemail_of_{name}example.com) app Litestar(route_handlers[get_person])本节的改动集中在三处新增两个导入DTOConfig和DataclassDTO均从litestar.dto导入。定义 DTO 类DTOConfig 用于配置 DTO。本例使用exclude{email}排除字段此外它还有大量其他配置选项本教程后续会逐一覆盖。DataclassDTO 是一个专门从 dataclass 生成 DTO 的工厂类同时它是一个typing.Generic泛型类接受类型参数。当我们提供类型参数时该类就变成泛型类的一个特化版本DataclassDTO[Person]即专门在Person实例与传输数据之间进行转换的 DTO 类型。路由处理器启用 DTO通过return_dtoReadDTO让 DTO 负责处理处理器返回值到响应数据的转换。注意并不必须通过子类化DataclassDTO来创建特化 DTO例如ReadDTO DataclassDTO[Person]同样是合法的特化 DTO。但子类化允许我们挂载配置对象config DTOConfig(...)同时完成类型特化因此教程采用子类方式。再次访问http://localhost:8000/person/peter响应中不再包含email字段至此我们已经成功隐藏了用户的邮箱地址。3. 排除嵌套模型的字段点分路径语法exclude选项不仅支持顶层字段还支持通过**点分路径dotted paths**定位到嵌套模型中的字段。例如exclude{a.b}将排除嵌套在a属性上的实例的b属性。我们为模型增加一个与Person关联的Address模型。完整代码见 nested_exclude.pyfrom __future__ import annotations from dataclasses import dataclass from litestar import Litestar, get from litestar.dto import DataclassDTO, DTOConfig from litestar.params import FromPath dataclass class Address: street: str city: str country: str dataclass class Person: name: str age: int email: str address: Address class ReadDTO(DataclassDTO[Person]): config DTOConfig(exclude{email, address.street}) get(/person/{name:str}, return_dtoReadDTO, sync_to_threadFalse) def get_person(name: FromPath[str]) - Person: # Your logic to retrieve the person goes here # For demonstration purposes, a placeholder Person instance is returned address Address(street123 Main St, cityCityville, countryCountryland) return Person(namename, age30, emailfemail_of_{name}example.com, addressaddress) app Litestar(route_handlers[get_person])Address模型有三个属性street、city、countryPerson模型新增了address属性。ReadDTO使用点分路径address.street排除了嵌套Address模型中的street字段。调用处理器后可以看到响应中address对象保留了city与country但street不再出现4. 排除集合中嵌套模型的字段类型参数索引语法在 Python 中泛型类型可以接受一个或多个类型参数方括号中的类型典型场景是表示某种类型的集合例如List[Person]List是泛型容器类型Person特化了集合中元素的类型。对于一个拥有任意数量类型参数的泛型类型例如GenericType[Type0, Type1, ..., TypeN]我们使用类型参数的索引来指明排除操作针对的是哪个类型a.0.b排除a的第一个类型参数Type0中实例的b字段a.1.b排除a的第二个类型参数Type1中实例的b字段依此类推。下面我们为Person模型增加一个自引用self-referencing的children关系。完整代码见 nested_collection_exclude.pyfrom __future__ import annotations from dataclasses import dataclass from litestar import Litestar, get from litestar.dto import DataclassDTO, DTOConfig from litestar.params import FromPath dataclass class Address: street: str city: str country: str dataclass class Person: name: str age: int email: str address: Address children: list[Person] class ReadDTO(DataclassDTO[Person]): config DTOConfig(exclude{email, address.street, children.0.email, children.0.address}) get(/person/{name:str}, return_dtoReadDTO, sync_to_threadFalse) def get_person(name: FromPath[str]) - Person: # Your logic to retrieve the person goes here # For demonstration purposes, a placeholder Person instance is returned address Address(street123 Main St, cityCityville, countryCountryland) child1 Person(nameChild1, age10, emailchild1example.com, addressaddress, children[]) child2 Person(nameChild2, age8, emailchild2example.com, addressaddress, children[]) return Person( namename, age30, emailfemail_of_{name}example.com, addressaddress, children[child1, child2], ) app Litestar(route_handlers[get_person])现在一个Person可以有多个children每个 child 又可以有多个 children层层嵌套。我们通过children.0.email和children.0.address显式排除了所有子Person的email与address字段children是list[Person]其第 0 个类型参数即Person。处理器中为Person增加了两个 child且每个 child 自身没有 children。输出如下子对象成功出现在响应中且它们的 email 与 address 都被排除。细心的读者可能会注意到我们并没有显式排除Person.children的children字段例如children.0.children但该字段并未出现在输出中。要理解原因我们来看下一节max_nested_depth配置。5. max_nested_depth控制嵌套深度上一节的现象是即便没有显式排除children.0.children每个嵌套Person的children集合也没有出现在响应里。按照未排除即应输出的直觉children集合中的每个Person应该有一个空的children集合——但事实并非如此。原因正是 DTOConfig.max_nested_depth 及其默认值1。max_nested_depth用于限制响应中包含的嵌套对象深度。在本例中Person有一个children集合集合元素是嵌套的Person对象——这算作 1 层嵌套深度Person.children中每个元素的children集合则处于第 2 层嵌套因此被max_nested_depth的默认值1排除掉了。下面修改脚本把max_nested_depth提升到2让孙级 children 也出现在响应中。完整代码见 max_nested_depth.py核心改动只有一处——DTO 配置class ReadDTO(DataclassDTO[Person]): config DTOConfig( exclude{email, address.street, children.0.email, children.0.address}, max_nested_depth2, )现在输出中可以看到那些空集合children: []本教程后续章节将恢复使用默认值1。从源码看max_nested_depth的默认值定义在 config.py 的DTOConfig数据类中max_nested_depth: int 1它的作用是在 DTO 后端生成传输模型时限制递归展开的层数既避免无限自引用模型导致递归失控也天然防止深层嵌套对象被意外暴露。6. 字段重命名显式声明与重命名策略字段在序列化时的名称可以通过两种方式改变显式声明新名称或声明重命名策略。6.1 显式重命名rename_fields我们可以通过DTOConfig.rename_fields属性显式重命名字段。它是一个字典键为原始字段名值为新字段名。下面的例子把address字段重命名为location完整代码见 explicit_field_renaming.pyclass ReadDTO(DataclassDTO[Person]): config DTOConfig( exclude{email, address.street, children.0.email, children.0.address}, rename_fields{address: location}, )响应中的address字段被重命名为location6.2 重命名策略rename_strategy除了逐字段显式重命名还可以使用字段重命名策略。策略通过DTOConfig.rename_strategy配置指定。Litestar 内置支持以下策略策略说明lower将字段名转换为小写upper将字段名转换为大写camel将字段名转换为驼峰式camel casepascal将字段名转换为帕斯卡式pascal case提示也可以自定义策略——向rename_strategy传入一个接收字段名并返回新字段名的可调用对象即可。把示例改为使用upper策略完整代码见 field_renaming_strategy.pyclass ReadDTO(DataclassDTO[Person]): config DTOConfig( exclude{email, address.street, children.0.email, children.0.address}, rename_strategyupper, )结果中所有字段名都被转换为大写从 config.py 的源码注释可以确认rename_fields中显式声明的字段不受rename_strategy影响Fields defined inrename_fieldsare ignored两者可以同时使用、各司其职。7. 接收数据从客户端控制入站数据到目前为止我们只处理了返回数据这一半。另一半是控制从客户端接收的数据。为了简化演示我们把数据模型缩减回只有name、age、email三个属性的Person。完整代码见 simple_receiving_data.pyfrom __future__ import annotations from dataclasses import dataclass from litestar import Litestar, post from litestar.dto import DataclassDTO, DTOConfig dataclass class Person: name: str age: int email: str class ReadDTO(DataclassDTO[Person]): config DTOConfig(exclude{email}) post(/person, return_dtoReadDTO, sync_to_threadFalse) def create_person(data: Person) - Person: # Logic for persisting the person goes here return data app Litestar(route_handlers[create_person])这里的要点与之前一样ReadDTO通过return_dto配置给处理器负责排除返回负载中的email字段处理器变成了 post() 处理器其函数签名同时声明了接受和返回Person实例Litestar 原生支持将请求负载解码为 Python dataclass所以本例即使不为入站数据定义 DTO 也能正常工作——入站方向的 DTO 是可选的。现在需要向服务器发送数据来测试程序可以使用 Postman 之类的工具或 Posting。下面是一个请求/响应负载示例8. 只读字段客户端永远不该提交的字段有些字段永远不应由客户端指定。例如创建新资源实例时模型的id字段应该由服务端生成而不是由客户端提交。下面我们给Person模型加上id字段并新建一个忽略id的WriteDTO。完整代码见 read_only_fields_error.pyfrom __future__ import annotations from dataclasses import dataclass from litestar import Litestar, post from litestar.dto import DataclassDTO, DTOConfig dataclass class Person: name: str age: int email: str id: int class ReadDTO(DataclassDTO[Person]): config DTOConfig(exclude{email}) class WriteDTO(DataclassDTO[Person]): config DTOConfig(exclude{id}) post(/person, dtoWriteDTO, return_dtoReadDTO, sync_to_threadFalse) def create_person(data: Person) - Person: # Logic for persisting the person goes here return data app Litestar(route_handlers[create_person])关键点WriteDTO被指示忽略id属性WriteDTO通过dtoWriteDTO关键字参数分配给处理器这意味着创建新Person实例时从客户端接收的任何数据中的id字段都会被忽略。当我们试图携带id字段创建新Person实例时会得到一个错误发生了什么DTO 试图构造Person模型实例但我们已经把id字段从接受的客户端数据中排除了。而id是Person模型必填字段模型构造函数因此抛错。解决这个问题有不止一种方式例如给id字段一个默认值并在处理器中覆盖默认值创建一个完全没有id字段的独立模型在处理器中把数据从该模型转移到Person模型。不过Litestar 内置了更优雅的方案DTOData。9. DTOData延迟实例化与数据访问有些时候数据不应当被立即解析成目标类的实例。上一节正是这样的例子当必填字段被客户端数据排除或缺失时立即实例化类必然报错。解决方案就是DTOData类型。完整代码见 dto_data.pyfrom __future__ import annotations from dataclasses import dataclass from litestar import Litestar, post from litestar.dto import DataclassDTO, DTOConfig, DTOData dataclass class Person: name: str age: int email: str id: int class ReadDTO(DataclassDTO[Person]): config DTOConfig(exclude{email}) class WriteDTO(DataclassDTO[Person]): config DTOConfig(exclude{id}) post(/person, dtoWriteDTO, return_dtoReadDTO, sync_to_threadFalse) def create_person(data: DTOData[Person]) - Person: # Logic for persisting the person goes here return data.create_instance(id1) app Litestar(route_handlers[create_person])要点拆解DTOData是一个数据容器既可以用于创建目标类实例也可以访问底层已解析、已校验的数据。本例从litestar.dto导入它处理器的数据参数类型从Person改为DTOData[Person]相应地注入到函数中的入站客户端数据将是一个DTOData实例在处理器内部我们为id字段生成一个值然后通过DTOData.create_instance()方法创建Person实例。从源码data_structures.py可见create_instance(**kwargs)会先把 DTO 校验后的数据self._data_as_builtins拷贝为字典再用传入的 kwargs 覆盖对应键最后交给 DTO 后端转换为目标类型实例——kwargs 的优先级高于 DTO 校验数据。应用恢复到正常工作状态技巧要为嵌套属性提供值可以使用双下划线语法作为create_instance()的关键字参数。例如address__id1会设置所创建实例的address属性的id。这一机制在源码中由_set_nested_dict_value()实现——它按__切分键名并递归写入嵌套字典。DTOData还有其他实用的方法我们将在下一节更新实例中看到。10. 更新实例PUT 与 PATCH本节展示如何使用DTOData更新已存在的实例。10.1 PUT 处理器全量替换语义PUT 请求的特征是要求提交完整的数据模型才能进行更新。完整代码见 put_handlers.pyfrom __future__ import annotations from dataclasses import dataclass from litestar import Litestar, put from litestar.dto import DataclassDTO, DTOConfig, DTOData from litestar.params import FromPath dataclass class Person: name: str age: int email: str id: int class ReadDTO(DataclassDTO[Person]): config DTOConfig(exclude{email}) class WriteDTO(DataclassDTO[Person]): config DTOConfig(exclude{id}) put(/person/{person_id:int}, dtoWriteDTO, return_dtoReadDTO, sync_to_threadFalse) def update_person(person_id: FromPath[int], data: DTOData[Person]) - Person: # Usually the Person would be retrieved from a database person Person(idperson_id, nameJohn, age50, emailemail_of_johnexample.com) return data.update_instance(person) app Litestar(route_handlers[update_person])要点脚本定义了一个路径为/person/{person_id:int}的 PUT 处理器路由参数person_id指明要更新哪个 person处理器中先创建一个Person实例模拟数据库查询结果然后把它传给DTOData.update_instance()方法该方法返回被修改后的同一个实例。从源码data_structures.py看update_instance(instance, **kwargs)的逻辑是把 DTO 校验数据与 kwargs 合并然后对每个键执行setattr(instance, k, v)原地修改并返回实例。调用效果10.2 PATCH 处理器部分更新语义与 PUT 要求提交整个数据模型不同PATCH 请求允许只提交数据模型属性的任意子集来进行更新。完整代码见 patch_handlers.pyfrom __future__ import annotations from dataclasses import dataclass from litestar import Litestar, patch from litestar.dto import DataclassDTO, DTOConfig, DTOData from litestar.params import FromPath dataclass class Person: name: str age: int email: str id: int class ReadDTO(DataclassDTO[Person]): config DTOConfig(exclude{email}) class PatchDTO(DataclassDTO[Person]): config DTOConfig(exclude{id}, partialTrue) patch(/person/{person_id:int}, dtoPatchDTO, return_dtoReadDTO, sync_to_threadFalse) def update_person(person_id: FromPath[int], data: DTOData[Person]) - Person: # Usually the Person would be retrieved from a database person Person(idperson_id, nameJohn, age50, emailemail_of_johnexample.com) return data.update_instance(person) app Litestar(route_handlers[update_person])改动要点处理器从put改为 patch() 处理器引入PatchDTO类配置与WriteDTO类似排除id但额外设置了partialTrue。该设置允许对资源进行部分更新——即客户端提交的字段中缺失的属性将被视为不更新而不是置空或报错。演示效果11. 在应用分层上声明 DTO从处理器到 Controller到目前为止DTO 都是逐个处理器声明的。真实应用往往有多个处理器让我们先看一个声明了多个处理器的脚本。完整代码见 multiple_handlers.pyfrom __future__ import annotations from dataclasses import dataclass from litestar import Litestar, patch, post, put from litestar.dto import DataclassDTO, DTOConfig, DTOData from litestar.params import FromPath dataclass class Person: name: str age: int email: str id: int class ReadDTO(DataclassDTO[Person]): config DTOConfig(exclude{email}) class WriteDTO(DataclassDTO[Person]): config DTOConfig(exclude{id}) class PatchDTO(DataclassDTO[Person]): config DTOConfig(exclude{id}, partialTrue) post(/person, dtoWriteDTO, return_dtoReadDTO, sync_to_threadFalse) def create_person(data: DTOData[Person]) - Person: # Logic for persisting the person goes here return data.create_instance(id1) put(/person/{person_id:int}, dtoWriteDTO, return_dtoReadDTO, sync_to_threadFalse) def update_person(person_id: FromPath[int], data: DTOData[Person]) - Person: # Usually the Person would be retrieved from a database person Person(idperson_id, nameJohn, age50, emailemail_of_johnexample.com) return data.update_instance(person) patch(/person/{person_id:int}, dtoPatchDTO, return_dtoReadDTO, sync_to_threadFalse) def patch_person(person_id: FromPath[int], data: DTOData[Person]) - Person: # Usually the Person would be retrieved from a database person Person(idperson_id, nameJohn, age50, emailemail_of_johnexample.com) return data.update_instance(person) app Litestar(route_handlers[create_person, update_person, patch_person])可以看到dtoWriteDTO, return_dtoReadDTO在三个处理器上重复出现。DTO 可以定义在应用的任何分层layer上这给了我们整理代码的机会——把处理器搬进一个 Controller并在 Controller 层定义 DTO。完整代码见 controller.pyfrom __future__ import annotations from dataclasses import dataclass from litestar import Controller, Litestar, patch, post, put from litestar.dto import DataclassDTO, DTOConfig, DTOData from litestar.params import FromPath dataclass class Person: name: str age: int email: str id: int class ReadDTO(DataclassDTO[Person]): config DTOConfig(exclude{email}) class WriteDTO(DataclassDTO[Person]): config DTOConfig(exclude{id}) class PatchDTO(DataclassDTO[Person]): config DTOConfig(exclude{id}, partialTrue) class PersonController(Controller): dto WriteDTO return_dto ReadDTO post(/person, sync_to_threadFalse) def create_person(self, data: DTOData[Person]) - Person: # Logic for persisting the person goes here return data.create_instance(id1) put(/person/{person_id:int}, sync_to_threadFalse) def update_person(self, person_id: FromPath[int], data: DTOData[Person]) - Person: # Usually the Person would be retrieved from a database person Person(idperson_id, nameJohn, age50, emailemail_of_johnexample.com) return data.update_instance(person) patch(/person/{person_id:int}, dtoPatchDTO, sync_to_threadFalse) def patch_person(self, person_id: FromPath[int], data: DTOData[Person]) - Person: # Usually the Person would be retrieved from a database person Person(idperson_id, nameJohn, age50, emailemail_of_johnexample.com) return data.update_instance(person) app Litestar(route_handlers[PersonController])对比可以看出之前的脚本为每条路由定义独立的处理器函数新脚本把这些路由组织进PersonController类从而把公共配置上移到 Controller 层在PersonController类上同时定义了dto WriteDTO与return_dto ReadDTO无需再在每个处理器上重复声明我们仍然在patch_person处理器上直接定义dtoPatchDTO用于覆盖 Controller 层级的dto设置——这正是 Litestar 分层配置的覆盖机制越内层的声明优先级越高。12. DTOConfig 全参数一览源码级本教程使用的所有配置都来自 DTOConfig。结合源码完整的可配置项如下配置项默认值说明exclude: set[str]set()显式排除字段。字段名为点分路径如address.street。指定exclude时未列出的字段默认包含。与include互斥同时指定会抛出ImproperlyConfiguredExceptioninclude: set[str]set()显式包含字段白名单模式。指定include时未列出的字段默认排除。与exclude互斥rename_fields: dict[str, str]dict()字段名到新名称的映射用于显式重命名rename_strategyNone重命名策略内置upper、lower、camel、pascal或传入自定义可调用对象。rename_fields中声明的字段不受其影响max_nested_depth: int1允许数据传输的嵌套最大深度用于防止深层/递归模型被无限展开partial: boolFalse是否允许传输部分数据PATCH 语义underscore_fields_private: boolTrue以下划线开头的字段视为私有字段默认排除在数据传输之外experimental_codegen_backend: bool \| NoneNone是否启用实验性代码生成后端forbid_unknown_fields: boolFalse原始数据中出现模型未定义的字段时是否抛出异常exclude与include的互斥校验在DTOConfig.__post_init__中强制执行见 config.py这一点从源码可以直接确认。深入阅读完整示例代码目录docs/examples/data_transfer_objects/factory/tutorialDTO 工厂更多用法白名单include、私有字段underscore_fields_private、未知字段处理等docs/examples/data_transfer_objects/factoryAPI 参考DTO 配置、DTO 数据结构、DataclassDTO、DTO 基类核心源码litestar/dto/config.py、litestar/dto/data_structures.py、litestar/dto/dataclass_dto.py单元测试可验证create_instance的双下划线嵌套参数等行为tests/unit/test_dto/test_factory/test_integration.py、tests/unit/test_dto/test_config.pyDTO 的进阶使用DTO 与请求/响应集成、自定义类型docs/usage/dto【免费下载链接】litestarLight, flexible and extensible ASGI framework | Built to scale项目地址: https://gitcode.com/GitHub_Trending/li/litestar创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考