重新設計公司的 API framework (3) - 實作細節

系列文章

前篇說到介面設計，這篇會開始探討實作問題。不過還是會盡量表述重點，避免通篇只是 copy paste 程式碼。

架構設計
#

先回顧一下上篇的 endpoint spec

@api.get("/user", style=WebStyle(error_map={UserNotFound: PlainText("user not found")}))
def get_user(query: UserId) -> Result[User, UserNotFound]: ...

可以發現，endpoint 的宣告資訊散落在 api.get decorator 和 function annotation 上。必須組合兩者才能建立完整的 endpoint spec。而有了這份 endpoint spec，才有辦法處理 HTTP request，以及產出 OpenAPI 文件。

因此我覺得應該這樣規劃

建模 endpoint spec。
實作 decorator 解析 endpoint 的宣告，並建立對應的 endpoint spec。
實作 runtime interpreter，在收到 HTTP request 時，根據 endpoint spec 處理請求。
實作 OpenAPI interpreter，根據 endpoint spec 產出 OpenAPI 文件。

注意：這邊要避免把 HTTP 處理邏輯直接往 decorator 裡面塞。這會讓 decorator 要同時負責解析、處理 request、甚至產生 OpenAPI，只會讓程式變成難以控管的大雜燴。讓解析 endpoint 和實際執行行為分開，會讓職責分離得更明確。

另外，先蒐集資訊也有額外的好處

可以做簡易防呆，例如 error_map 裡面填了一個未宣告在 errors 中的錯誤，可以在解析階段直接 raise exception 提醒，不用等到 runtime 才發現問題。
兩個 interpreter 都使用同一份 endpoint spec，就能減少 runtime 行為和 OpenAPI 文件之間的飄移。

建模 endpoint spec
#

接下來會看如何建模 endpoint spec，如果建模太爛，後面的 interpreter 就會寫得很辛苦。

直覺上來說就是宣告所有欄位，假設如下

class BaseError(Exception):
    code: str

class PlainText:
    text: str

class JsonCode:
    code: str

class WebStyle:
    error_map: dict[type, PlainText | JsonCode]

# 省略 AppStyle / ApiStyle 的宣告
Style = WebStyle | AppStyle | ApiStyle


class EndpointSpec:
    method: Literal["GET", "POST", "PUT", "DELETE"]
    path: str
    query: type | None
    body: type | None
    response: type
    errors: list[type[BaseError]]
    style: Style

像最開始舉例的 endpoint，實際資料就會像這樣

EndpointSpec(
    method="GET",
    path="/user",
    query=UserId,
    body=None,
    response=User,
    errors=[UserNotFound],
    style=WebStyle(
        error_map={UserNotFound: PlainText("user not found")}
    ),
)

看上去還不錯，但前篇有提到相容性，因此我們需要考量另一種 endpoint 形狀：

@api.legacy(  # get 改成 legacy
    "/user", style=WebStyle(error_map={UserNotFound: PlainText("user not found")})
)
def get_user(
    legacy: UserId,
) -> Result[User, UserNotFound]:  # query parameter 改成 legacy
    ...

有兩個更動

decorator 改為 legacy：代表 endpoint 預設同時接受 GET 和 POST。
function parameter 改為 legacy：代表資料來源不分 query parameter 和 body。

這邊為了簡化說明，先不讓 user 可以指定 primary HTTP method 和 primary input。

要紀錄 legacy，可以按照剛剛的邏輯，有什麼欄位加什麼，假設如下

class EndpointSpec:
    ...  # 省略剛剛提過的欄位
    secondary_method: Literal["GET", "POST", "PUT", "DELETE"] | None
    legacy: type | None

但如果像上面這樣建模，該怎麼撰寫 interpreter？試想

當 HTTP request 進來，找到符合的 endpoint spec，接下來發現此 endpoint spec 的 legacy 和 body 都有值。此時該報錯還是該把值丟進去？
當我要產生 OpenAPI，那我應該要把 legacy 和 body 的欄位合併，產生一個新的 model class，還是說就顯示其中一個就好？如果選擇合併，那當欄位衝突時該怎麼辦？

一個簡單的解決方案是在建立 EndpointSpec 的時候，發現衝突的情況就 raise exception。
這個做法只對了一半，因為即使寫了 raise exception，interpreter 還是要知道這個假設：「legacy 和 body 不會同時出現。」
而接手維護的人跟當初撰寫程式的人可能不是同一個，因此當他處理 interpreter 時，為了確認上述假設，還是得回頭過來理解 endpoint spec 的產生過程。

更好的方法是改進建模，讓錯誤的狀態無法被表示。可以改成像下方的宣告

HTTP_METHODS = Literal["GET", "POST", "PUT", "DELETE"]


class DirectRequest:
    method: HTTP_METHODS
    query: type | None = None
    body: type | None = None

class LegacyRequest:
    # 不需要再紀錄 method，因為 legacy endpoint 固定同時接受 GET 和 POST
    legacy: type | None = None

class EndpointSpec:
    path: str
    request: DirectRequest | LegacyRequest
    response: type
    ...  # 下面省略

這樣一來，當 interpreter 收到 HTTP request 時，endpoint spec 中的 request 類型只有可能是 DirectRequest 或 LegacyRequest 二選一。
此外，也可以保證只有 legacy endpoint 可以掛上 legacy 輸入參數。在 LegacyRequest 這個分支中，query 和 body 都是不存在的欄位，也不會有任何誤用的可能。

目前的建模其實仍有未表示好的地方，以錯誤風格來說

風格	成功回應	失敗回應
Web	`200`，可能是純文字或 json	`4xx`，可能是純文字或 json
App	`200`，json	`4xx`，json
API	`200`，json	`200`，json

這是不同的模型，因此 WebStyle, AppStyle 和 ApiStyle 裡的 error_map 模型顯然也不是同一種。
但我累了，就交給大家自行想像該如何建模了。

解析 endpoint 與 decorator
#

雖然 function annotation 解析邏輯我當初是靠老師傅手藝手刻，但以現今 AI 時代來說，直接請 AI 代勞是又快又好。因此就不再贅述解析 function annotation 的流程，只要記得在發現 endpoint 有問題時 raise exception 就好。

class ApiRouter:
    def get(self, path: str, style: Style):
        def decorator(handler):
            # 省略一堆解析 function annotation 的操作
            ...

            # 針對不合法的狀況丟出 exception
            if legacy is not None:
                raise ValueError("GET endpoint 不可以使用 legacy 輸入參數")
            if body is not None:
                raise ValueError("GET endpoint 不可以宣告 body")
            if not set(style.error_map).issubset(set(errors)):
                raise ValueError("error_map 含了未宣告在 errors 中的 error")

            # 省略註冊 endpoint，建構 endpoint spec 等操作
            ...

            return handler

        return decorator

    # 省略 post, legacy 等 method
    ...

實際使用起來就會像剛剛一樣

api = ApiRouter()

@api.get("/user", style=WebStyle(error_map={UserNotFound: PlainText("user not found")}))
def get_user(query: UserId) -> Result[User, UserNotFound]: ...

這個防護在大多情況下已經夠用了，不過還可以更好，因為這邊的裝飾對象非常明確。
以 get decorator 為例，他裝飾的 handler function：

input 可以接受 query parameter，也可以什麼都不填
return value 可以是一個能轉換成 json 的值 (僅成功路徑)，也可以是 Result。

因此我們可以建立一些 protocol 來描述可能出現的 input / output，然後限制 decorator 只能裝飾這些特定的 handler function。
如此一來，當 decorator 被誤用的時候，不用跑程式就可以收到 linter 警告。有助於加速開發。

# JsonSerializable 代表可以轉換成 json object 或 json array 的物件
# 實務上我使用 Pydantic BaseModel 實作，這裡先以抽象名稱表示
Returnable = JsonSerializable | Result

# 第一種類型，無任何參數的 handler function
class _NoSlot(Protocol):
    def __call__(self) -> Returnable: ...

# 第二種類型，僅有 query 參數的 handler function
class _QueryOnly(Protocol):
    def __call__(self, query) -> Returnable: ...

# get endpoint 接受上述兩種 handler function
GetHandler = TypeVar(
    "GetHandler",
    bound=_NoSlot | _QueryOnly,
)

class ApiRouter:
    def get(
        ... # 省略參數
    ) -> Callable[[GetHandler], GetHandler]:
        def decorator(handler: GetHandler) -> GetHandler:
            # 省略實作
            ...
            return handler
        return decorator

使用時

@api.get("/user", style=WebStyle())
# 用 body 會出現紅線
def get_user(body: UserId) -> Result[User, UserNotFound]: ...


@api.get("/user", style=WebStyle())
# int 雖然是合法的 json value，但不是本 framework 要求的 json 結構。
# 因此放在 return type 會出紅線。
def get_user(query: UserId) -> int: ...

legacy 也可以如法炮制，限制 legacy decorator 只能使用 legacy 輸入參數，盡可能將錯誤扼殺在源頭，不要等到執行了才發現有問題。

另外，這邊仍然留有漏洞：handler function 並沒有限制 query、body、legacy 等 parameter 的 type，也沒有限制 Result 的 success type 必須是 JsonSerializable、error type 必須是 BaseError。
這是可以做限制的，但要解決這個問題就必須提到 type variance，複雜程度可以直接開一篇新的文章，因此就先跳過。

Runtime interpreter
#

辛苦建立 endpoint spec 後，終於可以開始解決 runtime 行為。前幾篇有說到，底層 HTTP 函式庫用的是 Python Werkzeug，所以實際上就是找到 Werkzeug 解析完 HTTP request 要 dispatch 給 handler function 的那瞬間，將 interpreter 的行為插入其中。此時 interpreter 應該會收到幾個參數

Werkzeug 解析過的 raw input 資料（如 query parameter、form data、body 等）
原始 handler function
對應的 endpoint spec

而 runtime interpreter 具體來說要做下列事情

Input validation
Response serialization
Error handling

接下來會逐一介紹。

Input validation
#

將 Werkzeug 的 raw input 轉換成 handler function parameter 所定義的 type instance。第一篇提到，以前的人都等 HTTP request 進來之後，才在 endpoint 內自行處理 int / str，或一些非常基礎的格式/長度檢查。但這種 input validation 應該是非常常見且重複的行為，對於完全錯誤的 input，應該在 framework 層就直接攔掉，不要讓他有機會進到核心程式當中。
如果發現轉不成功，在這一步就該直接處理錯誤。

def runtime(werkzeug_request, handler, endpoint_spec):
    try:
        # 先判定 request 類型
        if isinstance(endpoint_spec.request, DirectRequest):
            query = None
            if cls := endpoint_spec.request.query:
                # 假設這邊是使用 validation library，將資料轉換成 instance
                query = convert(cls, **werkzeug_request.query)
            ...  # 省略 body
        else:
            # 這裡是 legacy request 類型
            legacy = None
            if cls := endpoint_spec.request.legacy:
                # legacy 的狀況，原始 query 和 body 都要倒進去
                legacy = convert(cls, **werkzeug_request.query, **werkzeug_request.body)
    except Exception:
        return validation_error()  # 驗證時期就失敗，直接回傳

Response serialization
#

將 handler function 所回傳的 instance 依照 API style 轉換成真正的 response body，別讓使用者在每個 handler function 裡不斷呼叫 json.dumps。
成功回傳值和錯誤回傳值的處理方式不同，因此分開進行序列化。

# 拿來裝回傳資料，一個 status + 一個 response body
# 實際上還可能有 media-type（因為 Web style 會輸出 text/plain，不過這邊就先省略）
class Response:
    status: int
    body: str

def runtime(werkzeug_request, handler, endpoint_spec):
    ... # 省略 input validation

    # 這邊要判定到底是哪種 request 類型，以及有什麼 input 參數可以傳
    # 不過我偷懶就先當作傳 query
    result = handler(query)

    # 根據不同 API style 輸出不同資料
    match endpoint_spec.style:
        case WebStyle():
            match result:
                case Ok(value):
                    # 成功的話，直接序列化回傳值
                    # 這邊假設 json.dumps 可以序列化 instance，實際上請根據自己的 library 做調整
                    return Response(200, json.dumps(value)) 
                case Err(err):
                    # 錯誤的話，先透過 error_map 轉換成對應的 response body
                    mapped = endpoint_spec.style.error_map.get(type(err))
                    # 再根據 response body 的類型進行序列化
                    if mapped is None:
                        # 預設格式
                        body = json.dumps({"code": err.code})
                    elif isinstance(mapped, PlainText):
                        # PlainText mapping 直接回傳文字
                        body = mapped.text
                    elif isinstance(mapped, JsonCode):
                        # JsonCode mapping 將 code map 成另一個 code
                        body = json.dumps({"code": mapped.code})
                    return Response(400, body)
        # 省略 AppStyle 和 ApiStyle 的處理
        ...

code 很長，但實際上做的事情就是

把已驗證資料傳入 handler function
根據 API style，將回傳值解釋成對應樣子

Error handling
#

同時也不要忘記錯誤處理，紀錄在案的 domain error 已經被一般路徑所處理。
但類似除以0，db 連不上，存取 array index 時超過範圍等真正的程式錯誤並沒有被處理。
這裡就是最佳的處理時機，用 try/except 捕捉任何未知的 exception。

這些未知的狀況應該

回傳 500，並用 Prometheus 等觀測工具紀錄，如果有超出預期的 500 比例，很可能是程式出了差錯。
未知的 exception 應該全部被 log 下來，這樣可以在 Prometheus 發出 alert 之後回頭查看問題。

def runtime(werkzeug_request, handler, endpoint_spec):
    ...  # 省略 input validation

    try:
        result = handler(query)
    except Exception:
        # Python 的 logger.exception 會自動紀錄 stack trace
        logger.exception("Unknown exception")
        return Response(500, "Unknown exception")

    # 省略輸出
    ...

OpenAPI interpreter
#

相較於 runtime interpreter，其實沒什麼值得說明的事。
我用的是 openapi-pydantic 套件，然後限制 request/response model 都只能使用 Pydantic BaseModel。硬要說的話就是還用到了 Pydantic 的動態 model 產生方法吧，以 API style 來說，回傳值一定跟當初定義的 response 長相不同。因此需要用點動態魔法創造新的 schema。不過這也是叫 AI 做可以做得很好的東西了，不需要像我兩年前一樣手刻。最後就是把 model class 傳進函式庫裡面就差不多了。

總結
#

到目前為止，就把整體實作流程說明完畢了。
我盡可能把 scope 減小，狀況單純化，以避免文章變成敝司的例外狀況建模大全。
在描述實作的時候也省略了不少東西，不過我覺得這部份讀者應該是能自行想像的。
如果覺得有某些地方有疑義的話，永遠記得：endpoint spec 越精確，interpreter 寫起來就越輕鬆，疑惑的源頭說不定是建模錯誤。即使是複雜的模型，如果以正確的角度切入的話，看上去也能變得單純。

架構設計#

建模 endpoint spec#

解析 endpoint 與 decorator#

Runtime interpreter#

Input validation#

Response serialization#

Error handling#

OpenAPI interpreter#

總結#