幫助我們為全公司范圍內的新 API 奠定基礎是令人興奮的。 2022 年初，我所在的團隊為我們的新 API 開發了概念驗證并建立了標準。 快進到今天，您會發現跨越 12 個產品領域的 150 多個端點每天處理數百萬個請求。 在這篇文章中，我將深入探討 API 的一個內部方面：數據傳輸對象 (DTO)。 我將討論為什么我們選擇 attrs 以及我們如何使用它。 我還將展示我們如何為開發人員標準化 API 實現流程，包括端點的版本控制。

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

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

設置場景

我們的 API 符合 JSON:API 規范。 API 團隊的 Chad Furman 寫了一篇很棒的文章，介紹了我們為什么選擇 JSON:API 以及我們如何使用它。 我們的實現是在 Python/ target=_blank class=infotextkey>Python 中使用 Django Rest Framework (DRF)。 我們利用 DRF 的可組合性和靈活性來定制 API 中的各種組件。

大致來說，實現如下：

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

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

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

使用數據傳輸對象 (DTO)

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

缺乏結構：詞典松散。 它們的外觀沒有界限。 很容易犯錯別字、數值多余、數值不足等錯誤。

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

從根本上講，DTO 是僅封裝數據的對象——它們內部幾乎沒有行為（最多是序列化邏輯）。 這些也稱為純數據類或數據類。 此類（或一般類）在實例化期間強制執行嚴格的模式。 也可以輕松地實現這些來實例化不可變對象。 另外，正如我們稍后將看到的，DTO 允許向屬性添加類型提示，這極大地提高了代碼的可讀性。

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

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

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

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

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

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

Klaviyo API DTO

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

在揭開該工具之前，讓我們看一個簡單的示例。 想象一下圖書館中有一個用于圖書的簡單 API。 此 API 的用戶希望使用搜索參數來查詢圖書。 以下是我們的一位開發人員如何編寫 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

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

@api_dto 裝飾器

attrs 字段包裝器

用于請求驗證的 common_validators 模塊

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

我們的 @api_dto 裝飾器是用如下代碼實現的：

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 應被凍結，以在 API 請求中的整個生命周期中強制執行不變性。 這也使得對象可散列，這有利于緩存請求和響應。

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

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

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

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

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

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

__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 的請求查詢參數。

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

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

我們沒有依賴開發人員在這個字典中自由設置值，而是在 attrs 字段函數周圍添加了一個簡單的包裝器。 該包裝器提供了一個一致的接口，用于設置其他關鍵字參數，如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))

這以可預測、干凈且穩健的方式構建元數據。 該包裝器中有一些有趣的參數：

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

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

驗證工具箱

如前所述，我們使用 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_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)

這總結了（抱歉，我無法抗拒）如何為 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:
        ...

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

準備 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 對象是簡單的數據類：

@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 元類構建的。 元類讀取此全局注冊表，并將方法名稱的映射附加到所有端點修訂版，作為相應 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 解釋器細節，它使裝飾器與元類無縫工作，從而使此設置成為可能：

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

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

3. 基礎方法（來自BaseApiViewSet）根據版本頭查找要調用的方法

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

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

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

最后，序列化器將 JSON 構建到綁定到該修訂版的 DTO 中，并將其傳遞給函數，執行該函數并將響應序列化回 JSON！

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

序列化

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

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

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

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

結論

我所描述的摘要：

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

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

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

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

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

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