幫助我們?yōu)槿痉秶鷥?nèi)的新 API 奠定基礎是令人興奮的。 2022 年初，我所在的團隊為我們的新 API 開發(fā)了概念驗證并建立了標準。 快進到今天，您會發(fā)現(xiàn)跨越 12 個產(chǎn)品領域的 150 多個端點每天處理數(shù)百萬個請求。 在這篇文章中，我將深入探討 API 的一個內(nèi)部方面：數(shù)據(jù)傳輸對象 (DTO)。 我將討論為什么我們選擇 attrs 以及我們?nèi)绾问褂盟?我還將展示我們?nèi)绾螢殚_發(fā)人員標準化 API 實現(xiàn)流程，包括端點的版本控制。

API 工作非常艱巨，涉及多個工程和產(chǎn)品團隊。 這需要討論、技術(shù)設計文件，當然還有一些自行車脫落！ 早期設計文檔中的這句話捕捉到了這一愿景：

“Klaviyo 值得擁有一個長期、一致且靈活的 API，它可以在未來幾年為 Klaviyo 內(nèi)部和外部的開發(fā)人員提供服務，同時最大限度地減少我們內(nèi)部開發(fā)人員的運營開銷，并最大限度地提高我們外部開發(fā)人員的一致性和可用性。”

設置場景

我們的 API 符合 JSON:API 規(guī)范。 API 團隊的 Chad Furman 寫了一篇很棒的文章，介紹了我們?yōu)槭裁催x擇 JSON:API 以及我們?nèi)绾问褂盟?我們的實現(xiàn)是在 Python/ target=_blank class=infotextkey>Python 中使用 Django Rest Framework (DRF)。 我們利用 DRF 的可組合性和靈活性來定制 API 中的各種組件。

大致來說，實現(xiàn)如下：

API 路由注冊在 Router 對象上，該對象將傳入請求（正文、查詢參數(shù)、標頭等）分派到相應的 ViewSet 類。

通過在 ViewSet 類上配置自定義身份驗證、許可和速率限制邏輯，將它們插入到 DRF 中。 這些由 DRF 在傳入請求時調(diào)用。

ViewSet 類上實現(xiàn)的不同 HTTP 方法（GET、POST、PATCH 等）通過調(diào)用內(nèi)部服務來處理傳入請求。 這通常會通過適配器層來處理各個方向的有效負載，最終返回 HTTP 響應。

使用數(shù)據(jù)傳輸對象 (DTO)

在 Python 中，使用普通的舊字典很容易表示鍵值數(shù)據(jù)（例如 JSON 有效負載），但這種便利性可能會代價高昂：

缺乏結(jié)構(gòu)：詞典松散。 它們的外觀沒有界限。 很容易犯錯別字、數(shù)值多余、數(shù)值不足等錯誤。

可變性：Python 中的字典是可變的，如果您使用該語言一段時間，您已經(jīng)知道這可能會導致各種令人討厭的錯誤。

從根本上講，DTO 是僅封裝數(shù)據(jù)的對象——它們內(nèi)部幾乎沒有行為（最多是序列化邏輯）。 這些也稱為純數(shù)據(jù)類或數(shù)據(jù)類。 此類（或一般類）在實例化期間強制執(zhí)行嚴格的模式。 也可以輕松地實現(xiàn)這些來實例化不可變對象。 另外，正如我們稍后將看到的，DTO 允許向?qū)傩蕴砑宇愋吞崾?，這極大地提高了代碼的可讀性。

我們使用 DTO 來代表我們的 API 合約。 每個端點（HTTP 方法）都有一個關聯(lián)的入口 DTO（表示傳入請求的 JSON 正文以及查詢參數(shù)）和一個相關的響應 DTO（表示返回到客戶端的 JSON 正文）。 例如，當使用我們的 API 創(chuàng)建目錄項時，請求和響應數(shù)據(jù)字典將被建模為 DTO。

我們不鼓勵使用可以跨多個端點重用的通用、稀疏實例化的 DTO。 盡管這增加了一些冗余，但它提供了清晰、嚴格的模式實施，并且還產(chǎn)生了模塊化設計，可以輕松獨立地對不同端點的合約進行版本控制。 此外，這種獨特的 DTO 端點綁定有助于簡化公共 API 文檔的自動生成。

當時，一些庫已經(jīng)提供了創(chuàng)建純數(shù)據(jù)類的出色解決方案，而開發(fā)人員無需編寫標準 Python 類通常所需的樣板代碼。 最流行的是：dataclasses、pydantic 和 attrs。 我不會詳細比較這三者，因為有很多文章（請參閱 Attrs、Dataclasses 和 Pydantic 以及為什么我使用 attrs 而不是 pydantic）。

在較高的層面上，第一個決定是在 attrs/dataclasses 和 pydantic 之間。 前兩個與 pydantic 相似但又截然不同。 Pydantic 主要是一個驗證庫而不是數(shù)據(jù)容器。 盡管在這里使用 pydantic 很誘人，因為它適合我們的用例，但我們主要出于性能原因決定不使用它。 我們的 DTO 需要在每個 API 請求的 API Web 層上同步實例化，因此每個潛在的性能瓶頸都很重要。 這篇博文對這些庫的性能進行了一些有趣的研究和基準測試。

我們選擇了 attrs，因為它高性能、功能豐富（與數(shù)據(jù)類相比）、靈活且易于使用。 另外，由于 attrs 不是標準庫的一部分（與數(shù)據(jù)類不同），合并新功能不需要 Python 版本升級。 就我個人而言，我真的很喜歡他們的裝飾器風格模式，而不是 pydantic 使用的繼承。 他們在哲學上更傾向于組合而不是繼承，這使得組合更加透明并且易于針對我們的用例進行定制。 attrs 將方法附加到類上，一旦類生成裝飾器執(zhí)行，它就是一個普通的舊 Python 類。

“它在運行時不執(zhí)行任何動態(tài)操作，因此運行時開銷為零。 這仍然是你的Class。 ” — attrs 文檔

Klaviyo API DTO

一般來說，在合理的情況下包裝第三方庫被認為是良好的做法。 首先，它有助于統(tǒng)一使用，例如具有許多選項的庫的所需設置和默認設置。 其次，它創(chuàng)建了一個應用普遍變革的中心點。 第三，它提供了抽象庫細節(jié)的機會，從而提供了將其替換為替代方案的靈活性，在這種情況下，包裝器充當適配器。 出于所有這些原因，我們將 attrs 包裝在為開發(fā)人員提供的工具中。

在揭開該工具之前，讓我們看一個簡單的示例。 想象一下圖書館中有一個用于圖書的簡單 API。 此 API 的用戶希望使用搜索參數(shù)來查詢圖書。 以下是我們的一位開發(fā)人員如何編寫 API 的查詢請求 DTO：

from App.views.apis.v3.dtos import api_dto, field
from app.views.apis.v3.validation import common_validators

@api_dto(ApiResourceEnum.BOOK, enable_boolean_filters=True)
class BookQueryDTO:
    id: str | None = None
    title: str | None = field(
        default=None, 
        external_desc="Title of the book you are querying for",
        example="Harry Potter and The Sorcerer's Stone", 
        filter_operators={FilterOperators.CONTAINS, FilterOperators.EQUALS}, 
        validator=common_validators.max_len(100)
        sortable=True
    )
    author_id: str | None = field(
        default=None, 
        filter_operators={FilterOperators.EQUALS}, 
    )
    page_cursor: str | None = None
    return_fields: list[str] | None = None
    sort: str | None = None

上面的示例將我們將在接下來的部分中討論的幾個重要部分結(jié)合在一起：

@api_dto 裝飾器

attrs 字段包裝器

用于請求驗證的 common_validators 模塊

我們的 @api_dto 裝飾器：包裝 attrs @define

我們的 @api_dto 裝飾器是用如下代碼實現(xiàn)的：

from attrs import define, resolve_types

def api_dto(
    resource: ApiResource,
    enable_boolean_filters: bool = False,
    non_dto_sort_fields: list | None = None,
    min_max_page_size: tuple[int | None, int | None] | None = None,
) -> Callable:

    # ... 
    # Arg validation
    # ...

    def inner(py_dto_cls: type) -> ApiDtoClass:
        generated_attr_dto = resolve_types(
            define(frozen=True, kw_only=True, auto_attribs=True)(py_dto_cls)
        )

        setattr(generated_attr_dto, "__api_dto__", True)
        setattr(generated_attr_dto, "__resource__", resource)
        setattr(generated_attr_dto, "__boolean_filters_enabled__", enable_boolean_filters)

        if non_dto_sort_fields:
            setattr(generated_attr_dto, "__non_dto_sort_fields__", non_dto_sort_fields)

        if min_max_page_size:
            setattr(generated_attr_dto, "__min_max_page_size__", min_max_page_size)

        # ...
        # More Validation to ensure proper setup 
        # ...

        return generated_attr_dto

    return inner

它將 attrs 定義裝飾器應用到傳入的類，并具有預定的配置：

freeze=True 這些 DTO 應被凍結(jié)，以在 API 請求中的整個生命周期中強制執(zhí)行不變性。 這也使得對象可散列，這有利于緩存請求和響應。

kw_only=True 由于這些 DTO 可能具有多個屬性，因此為了清楚起見，必須僅使用關鍵字參數(shù)來實例化這些屬性。

auto_attribs=True 這是 attrs 的一個很好的功能，它避免了將每個屬性分配給字段的需要。 它還強制執(zhí)行類型注釋。

這里一個更重要的細節(jié)是，define 裝飾器默認生成一個開槽類 (slots=True)，因此這些 DTO 的內(nèi)存占用較小，這是有助于擴展的又一個因素。

盡管 attrs 在定義裝飾器中還有其他幾個參數(shù)，但到目前為止我們的 API DTO 還不需要它們，而且我們的包裝器使我們的內(nèi)部開發(fā)人員不必考慮它們。

最后，我們在這個修飾類上解析_types()，以允許前向引用的字符串類型提示。 這可確保定義每個屬性的類型并準備好用于序列化/反序列化。

您可能已經(jīng)注意到，生成此類對象后，接下來的幾行會在類（而不是實例）上設置一些屬性值：

__resource__ 屬性引用 ApiResource 對象。該對象存儲此 DTO 建模的資源的類型等。 然后由序列化和文檔工具使用。 每個域都有一個枚舉來保存其所有 ApiResource 對象。 然后在裝飾器上提供枚舉，例如 ApiResourceEnum.BOOK。

__api_dto__ 是一個標志，指示此類是使用此裝飾器生成的。 這充當在 DTO 注冊表期間進行驗證的水印，以確保所有 API DTO 都是從此裝飾器生成的。

__boolean_filters_enabled__ 屬性是一個開關，允許使用 AND / OR / NOT 布爾運算符過濾 DTO 中的字段。

__non_dto_sort_fields__ 和 __min_max_page_size__ 幫助解析和處理此 DTO 的請求查詢參數(shù)。

我們的 API DTO 字段：包裝 attrs 字段

attrs 允許將元數(shù)據(jù)附加到屬性，事實證明，這非常漂亮！ 我們用它來存儲 API 不同部分使用的屬性信息：文檔、過濾、編輯等。

我們沒有依賴開發(fā)人員在這個字典中自由設置值，而是在 attrs 字段函數(shù)周圍添加了一個簡單的包裝器。 該包裝器提供了一個一致的接口，用于設置其他關鍵字參數(shù)，如filter_operators、sortable、external_desc 等（見下文）。 這是代表我們的字段包裝器的片段：

from attrs import field as attrs_field

def 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=(metadata or None))

這以可預測、干凈且穩(wěn)健的方式構(gòu)建元數(shù)據(jù)。 該包裝器中有一些有趣的參數(shù)：

filter_operators 用于在 API 請求中指定該字段可能的過濾運算符。 我們有自己的過濾語法（使用 pyparsing 實現(xiàn)），可以解析 JSON API 過濾器并使用此處指定的運算符驗證請求。 這個 kwarg 只是冰山一角，我認為我們的 API 過濾語法值得單獨寫一篇文章。

external_desc 和 example 字段由生成 OpenAPI 規(guī)范文檔的內(nèi)部工具使用。 這通過 DTO 代碼更改（我們的 API 合約）簡化了文檔更新。 開發(fā)人員只需使用此 kwarg 在 DTO 字段上配置新信息，文檔就會使用該信息進行更新！

驗證工具箱

如前所述，我們使用 DTO 來表示請求中的 JSON 正文。 我們添加了一個層，即使在無效負載進入內(nèi)部服務邊界之前，它也會給我們帶來拒絕無效負載的溫暖模糊感覺！

此驗證將在 API Web 服務器上同步進行，因此，我們需要謹慎對待這些 DTO 的驗證范圍。 例如，我們不想在這里進行數(shù)據(jù)庫調(diào)用； 這將發(fā)生在內(nèi)部服務邊界。 我們的想法是進行輕量級驗證，足以拒絕不必要地使用堆棧更深層次資源的不良有效負載。

attrs 通過將驗證器函數(shù)指定為字段上的 kwarg，可以輕松驗證這些數(shù)據(jù)類。 這些驗證器在對象實例化時運行（在本例中將原始 JSON 反序列化為請求 DTO）。 我們的內(nèi)部開發(fā)人員可以訪問這些驗證器的精益包裝器，以生成一致的錯誤消息。 使用裝飾器來定義錯誤消息，我們現(xiàn)在可以中繼回狀態(tài)為 400 的 HTTP 響應。 通常，我們會對代表請求的 DTO 添加嚴格的驗證，而不是對響應的 DTO 進行太多驗證。 這是因為我們可以控制后者的生成，并且可以使用自動化測試來確保正確性。

我們的 API 代碼庫中的 Python 模塊封裝了可供所有團隊使用的通用 DTO 驗證器。 在這些驗證器中，許多只是 attrs 驗證器的包裝器，而其他驗證器則是在這些驗證器的基礎上構(gòu)建的。 它們構(gòu)成了實現(xiàn) DTO 時使用的工具箱。 許多團隊最終編寫了自己的驗證器模塊，特定于他們的領域，并基于這些基本驗證器構(gòu)建。 如果驗證器足夠通用，足以對其他團隊有用，那么它就會進入基本驗證器模塊。

我們還有一個模塊，用于維護生成驗證器函數(shù)的 Python 閉包。 這里的想法是，有時不同的團隊可能最終會實現(xiàn)具有相同驗證邏輯的類似驗證器，只是不同的“參數(shù)”。 擁有這個模塊有助于消除冗余。 此閉包的一個簡單示例如下所示：

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_validator
    def generated_validator_fn(instance, attribute, value):
        if value % divisor != 0:
            raise ValueError(f"{value=} is not divisible by {divisor=}")

    return generated_validator_fn

# Example use:
# divisible_by_two_validator = divisible_by__validator_closure(2)

這總結(jié)了（抱歉，我無法抗拒）如何為 Klaviyo 的 API 創(chuàng)建 DTO。 JSON:API 關系也在這些 DTO 中建模，但為了簡潔起見，我們不會在本文中介紹它們。

DTO 和 ViewSet 元編程的注冊表

到目前為止，在這篇文章中，我們揭示了 DTO 在 API 中代表什么以及它們是如何以標準化方式創(chuàng)建的。 但是，每個版本的 API 端點如何知道要使用哪個 DTO？ 此外，一旦解決了這個問題，入站原始 JSON 如何轉(zhuǎn)換為該 DTO（其他方向也類似）？

為了回答這些問題，讓我們了解 ViewSet 類是如何實現(xiàn)和版本控制的。 使用上面的 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:
        ...

上面的示例中發(fā)生了一些有趣的事情，我們將看看它是如何在幕后引導的（包括使同一個類中可以具有相同名稱的方法而無需任何真正的重載的魔力） ）。

準備 Klaviyo API ViewSet 類大致分為三個步驟：

@api_revision 裝飾器將所有 ViewSet 的所有方法的全局注冊表填充到特定修訂版，并按類名鍵入。 例如：

{
    "BooksViewSet": {
        "list": [Revision(...), Revision(...)],
        "create": [Revision(...)],
    },
    "FooViewSet": {
        "list": [Revision(...), Revision(...), Revision(...)],
        "create": [Revision(...)],
        "retrieve": [Revision(...), Revision(...)],
        "update": [Revision(...)],
        "partial_update": [Revision(...)],
        "destroy": [Revision(...)],
    },
    "BarViewSet": {...}
}

上面注冊表中的 Revision 對象是簡單的數(shù)據(jù)類：

@dataclass
class Revision:
    """A single API method revision's information, defaults are set in the api_revision decorator"""
    revision_date: str
    func: Callable
    auto_deprecate: bool
    deprecation_date: str
    removal_date: str
    ingress_dto_cls: Type[API_DTO]
    egress_dto_cls: Type[API_DTO] = None

這意味著每個方法修訂版都有一個與其綁定的 DTO，并存儲在該全局注冊表中。

2. BaseApiViewSet 是使用 ApiV3Metaclass 元類構(gòu)建的。 元類讀取此全局注冊表，并將方法名稱的映射附加到所有端點修訂版，作為相應 ViewSet 上的類屬性。

元類大致如下所示：

class ApiV3Metaclass(type):
    def __new__(mcs, class_name, bases, attrs):
        # attributes to attach to the class
        attrs_to_build = dict()

        # collection of our api methods and revisions, we want to structure this data
        # to make incoming requests as fast as possible to route at runtime
        # { method_name -> [Revision(...), Revision(...), ...] }
        revision_list_by_viewset_method = defaultdict(list)

        # Create the revision methods on the class based on the revision fed into the
        # @api_revision decorator
        for (
            viewset_method_name,
            revisions,
        ) in _funcs_and_revisions_by_class_and_method[class_name].items():
            revision_list_by_viewset_method[viewset_method_name] = sorted(
                revisions,
                key=lambda revision: RevisionDate(revision.revision_date),
                reverse=True,
            )

        attrs_to_build["revisions_by_method"] = revision_list_by_viewset_method

 # ...         
        # More setup
 # ...

        return super(APIV3Metaclass, mcs).__new__(
            mcs, class_name, bases, attrs_to_build
        )

每個視圖集都有一個 revisions_by_method 屬性，如下所示：

{
  "list": [Revision(...), Revision(...), Revision(...)],
  "create": [Revision(...)],
  "retrieve": [Revision(...), Revision(...)],
  "update": [Revision(...)],
  "partial_update": [Revision(...)],
  "destroy": [Revision(...)],
}

有一個有趣的 Python 解釋器細節(jié)，它使裝飾器與元類無縫工作，從而使此設置成為可能：

在Python中，類主體在使用確定的元類設置類之前執(zhí)行。 這里有關于這個過程的更多細節(jié)，但對于我們的場景來說，這意味著裝飾器首先執(zhí)行（填充全局注冊表），然后執(zhí)行元類 __new__ 方法，該方法使用此全局注冊表創(chuàng)建一個類屬性，該屬性存儲修訂 方法。

修飾方法從未真正附加到類，而僅作為 Revision 對象中的引用存在。 這就是為什么可以有同名的方法！

3. 基礎方法（來自BaseApiViewSet）根據(jù)版本頭查找要調(diào)用的方法

BaseApiViewset 類包含所有 ViewSet 操作方法（列表、創(chuàng)建、檢索等）的簡單實現(xiàn)。 這個簡單的部分實際上是將所有這些組合在一起的：

（回顧）路由器將請求分派到相應的 ViewSet 類。 由于裝飾方法從未附加，因此存在的這些方法的唯一實現(xiàn)來自基本方法，該方法在此處被調(diào)用。

基本方法解析請求標頭以獲取所請求端點的修訂日期。 它從 ViewSet 類上的 revisions_by_method 查找中獲取特定的 Revision 對象。 回想一下，此 Revision 對象保存對端點特定版本的 DTO 和函數(shù)引用。

最后，序列化器將 JSON 構(gòu)建到綁定到該修訂版的 DTO 中，并將其傳遞給函數(shù)，執(zhí)行該函數(shù)并將響應序列化回 JSON！

所有 API ViewSet 都繼承自 BaseApiViewset 并使用此機制工作。

序列化

我們的 API 使用 cattrs 來完成與 JSON 的序列化/反序列化。 這是一個方便的 Python 庫，可以幫助構(gòu)建非結(jié)構(gòu)化數(shù)據(jù)（如字典），反之亦然。 該庫功能強大，提供多種可能的轉(zhuǎn)換。 （盡管這聽起來似乎沒什么大不了的，但我認為將原始值轉(zhuǎn)換為枚舉的能力非常方便且非常簡潔。）

cattrs 與 attrs 集成得很好，這使得它成為我們 API 的一個簡單選擇。 我還喜歡使用 ExceptionGroups 在 cattrs 中進行異常處理：它在序列化器層中很有用，我們需要精確地（外部或內(nèi)部）查明 DTO 未能創(chuàng)建的位置及其原因。

cattrs 的使用是 API 系統(tǒng)內(nèi)部的。 我們有一個特定的 APIConverter 類，其中注冊了一些默認掛鉤，還有一個注冊表，其他團隊可以在其中幫助處理特定的罕見邊緣情況。 默認鉤子的用途不僅僅是提供各種轉(zhuǎn)換。 在某些情況下，此處注冊的掛鉤可以充當特定類型的全局驗證器。 例如，日期和日期時間類型可以接受各種格式，并且此轉(zhuǎn)換器負責驗證這些格式（而不是每個 DTO 都必須驗證它），并在出現(xiàn)問題時引發(fā)驗證錯誤。

總的來說，我們對cattrs有一個普通的用法，并且事實證明它是有效的。

結(jié)論

我所描述的摘要：

我們的 API 使用 DTO 表示請求正文、查詢參數(shù)和響應。

我們包裝了 attrs 來簡化和標準化開發(fā)人員的使用。

我們采用裝飾器和元類等 Python 模式來大規(guī)模簡化實現(xiàn)、對 API 進行版本控制，并將 DTO 綁定到端點版本。

用 cattrs 補充 attrs 簡化了 DTO 的序列化和反序列化。

我們對 2022 年初做出的決定感到滿意。與每個端點直接接受和生成 JSON 或字典相比，我們基于 DTO 的方法有助于實現(xiàn)上面提到的愿景：

“Klaviyo 值得擁有一個長期、一致且靈活的 API，它可以在未來幾年為 Klaviyo 內(nèi)部和外部的開發(fā)人員提供服務，同時最大限度地減少我們內(nèi)部開發(fā)人員的運營開銷，并最大限度地提高我們外部開發(fā)人員的一致性和可用性。”