使用 *attr* DTO 为我们的 API 提供支持( 三 )


from attrs import field as attrs_fielddef field(*args,filter_operators: set[FilterOperators] | None = None,non_filterable: bool = False,sortable: bool = False,accept_multiple_query_param: bool = False,external_desc: str | None = None,example: Any | None = None,data_classification: DataClassification = DataClassification.DEFAULT,meta: bool = False,**kwargs,):# ...# Parse and validate args# ...# ...# Construct field metadata in a standardized fashion (fixed keys, internal to API machinery)# eg. metadata["__external_desc__"] = external_desc# ...return attrs_field(*args, **kwargs, metadata=https://www.isolves.com/it/cxkf/bk/2023-08-31/(metadata or None))这以可预测、干净且稳健的方式构建元数据 。该包装器中有一些有趣的参数:
filter_operators 用于在 API 请求中指定该字段可能的过滤运算符 。我们有自己的过滤语法(使用 pyparsing 实现),可以解析 JSON API 过滤器并使用此处指定的运算符验证请求 。这个 kwarg 只是冰山一角,我认为我们的 API 过滤语法值得单独写一篇文章 。
external_desc 和 example 字段由生成 OpenAPI 规范文档的内部工具使用 。这通过 DTO 代码更改(我们的 API 合约)简化了文档更新 。开发人员只需使用此 kwarg 在 DTO 字段上配置新信息,文档就会使用该信息进行更新!
验证工具箱 

使用 *attr* DTO 为我们的 API 提供支持

文章插图
 
如前所述,我们使用 DTO 来表示请求中的 JSON 正文 。我们添加了一个层,即使在无效负载进入内部服务边界之前,它也会给我们带来拒绝无效负载的温暖模糊感觉!
此验证将在 API Web 服务器上同步进行,因此,我们需要谨慎对待这些 DTO 的验证范围 。例如,我们不想在这里进行数据库调用; 这将发生在内部服务边界 。我们的想法是进行轻量级验证,足以拒绝不必要地使用堆栈更深层次资源的不良有效负载 。
attrs 通过将验证器函数指定为字段上的 kwarg,可以轻松验证这些数据类 。这些验证器在对象实例化时运行(在本例中将原始 JSON 反序列化为请求 DTO) 。我们的内部开发人员可以访问这些验证器的精益包装器,以生成一致的错误消息 。使用装饰器来定义错误消息,我们现在可以中继回状态为 400 的 HTTP 响应 。通常,我们会对代表请求的 DTO 添加严格的验证,而不是对响应的 DTO 进行太多验证 。这是因为我们可以控制后者的生成,并且可以使用自动化测试来确保正确性 。
我们的 API 代码库中的 Python 模块封装了可供所有团队使用的通用 DTO 验证器 。在这些验证器中,许多只是 attrs 验证器的包装器,而其他验证器则是在这些验证器的基础上构建的 。它们构成了实现 DTO 时使用的工具箱 。许多团队最终编写了自己的验证器模块,特定于他们的领域,并基于这些基本验证器构建 。如果验证器足够通用,足以对其他团队有用,那么它就会进入基本验证器模块 。
我们还有一个模块,用于维护生成验证器函数的 Python 闭包 。这里的想法是,有时不同的团队可能最终会实现具有相同验证逻辑的类似验证器,只是不同的“参数” 。拥有这个模块有助于消除冗余 。此闭包的一个简单示例如下所示:
def divisible_by__validator_closure(divisor: int) -> Callable:if not isinstance(divisor, int):raise ValueError(f"divisor must be of type int, got {type(divisor)}")if divisor == 0:raise ZeroDivisionError("Cannot use 0 as a divisor")@api_custom_validatordef generated_validator_fn(instance, attribute, value):if value % divisor != 0:raise ValueError(f"{value=https://www.isolves.com/it/cxkf/bk/2023-08-31/} is not divisible by {divisor=}")return generated_validator_fn# Example use:# divisible_by_two_validator = divisible_by__validator_closure(2)这总结了(抱歉,我无法抗拒)如何为 Klaviyo 的 API 创建 DTO 。JSON:API 关系也在这些 DTO 中建模,但为了简洁起见,我们不会在本文中介绍它们 。
DTO 和 ViewSet 元编程的注册表到目前为止,在这篇文章中,我们揭示了 DTO 在 API 中代表什么以及它们是如何以标准化方式创建的 。但是,每个版本的 API 端点如何知道要使用哪个 DTO? 此外,一旦解决了这个问题,入站原始 JSON 如何转换为该 DTO(其他方向也类似)?
为了回答这些问题,让我们了解 ViewSet 类是如何实现和版本控制的 。使用上面的 Books API 示例,Klaviyo API ViewSet 如下所示:
class BooksViewSet(BaseApiViewSet):@api_revision("2020-01-01",ingress_dto_type=BooksListQuery,egress_dto_type=BooksResponse,)def list(self, request: Request, request_dto: API_DTO) -> JsonApiResponse:...@api_revision("2023-06-01",ingress_dto_type=BooksListQuery,egress_dto_type=BooksResponse,)def list(self, request: Request, request_dto: API_DTO) -> JsonApiResponse:...@api_revision("2020-05-05",auto_deprecate=False,ingress_dto_type=BookCreateQuery,egress_dto_type=BookResponse,)def create(self, request, request_dto: API_DTO) -> JsonApiResponse:...


推荐阅读