跳至內容

OpenAPI 回呼

您可以建立一個帶有「路徑操作」的 API，該操作可以觸發對其他人（可能是與使用您的 API 的開發者相同）建立的「外部 API」的請求。

當您的 API 應用程式呼叫「外部 API」時發生的過程稱為「回呼」。因為外部開發人員編寫的軟體會向您的 API 發送請求，然後您的 API 會「回呼」，向「外部 API」（可能是由同一位開發人員建立的）發送請求。

在這種情況下，您可能想要記錄該外部 API 應該是什麼樣子的。它應該有什麼「路徑操作」、應該接受什麼主體、應該返回什麼響應等等。

具有回呼的應用程式

讓我們用一個例子來看這一切。

假設您開發了一個允許建立發票的應用程式。

這些發票將具有 id、title（可選）、customer 和 total。

您的 API 使用者（外部開發人員）將使用 POST 請求在您的 API 中建立發票。

然後您的 API 將（讓我們想像一下）

  • 將發票發送給外部開發人員的某些客戶。
  • 收款。
  • 將通知發送回 API 使用者（外部開發人員）。
    • 這將透過向該外部開發人員提供的某些「外部 API」發送 POST 請求（從您的 API）來完成（這就是「回呼」）。

一般的 FastAPI 應用程式

讓我們先看看在新增回呼之前，一般的 API 應用程式是什麼樣子的。

它將有一個「路徑操作」，該操作將接收 Invoice 主體和一個包含回呼 URL 的查詢參數 callback_url。

這部分很正常，大部分程式碼您可能已經很熟悉了

from typing import Union

from fastapi import APIRouter, FastAPI
from pydantic import BaseModel, HttpUrl

app = FastAPI()


class Invoice(BaseModel):
    id: str
    title: Union[str, None] = None
    customer: str
    total: float


class InvoiceEvent(BaseModel):
    description: str
    paid: bool


class InvoiceEventReceived(BaseModel):
    ok: bool


invoices_callback_router = APIRouter()


@invoices_callback_router.post(
    "{$callback_url}/invoices/{$request.body.id}", response_model=InvoiceEventReceived
)
def invoice_notification(body: InvoiceEvent):
    pass


@app.post("/invoices/", callbacks=invoices_callback_router.routes)
def create_invoice(invoice: Invoice, callback_url: Union[HttpUrl, None] = None):
    """
    Create an invoice.

    This will (let's imagine) let the API user (some external developer) create an
    invoice.

    And this path operation will:

    * Send the invoice to the client.
    * Collect the money from the client.
    * Send a notification back to the API user (the external developer), as a callback.
        * At this point is that the API will somehow send a POST request to the
            external API with the notification of the invoice event
            (e.g. "payment successful").
    """
    # Send the invoice, collect the money, send the notification (the callback)
    return {"msg": "Invoice received"}

提示

callback_url 查詢參數使用 Pydantic Url 類型。

唯一新的東西是作為「路徑操作裝飾器」參數的 callbacks=invoices_callback_router.routes。我們接下來會看到那是什麼。

記錄回呼

實際的回呼程式碼將很大程度上取決於您自己的 API 應用程式。

而且它可能會因應用程式而異。

它可能只是一兩行程式碼，例如

callback_url = "https://example.com/api/v1/invoices/events/"
httpx.post(callback_url, json={"description": "Invoice paid", "paid": True})

但回呼中最重要的部分可能是確保您的 API 使用者（外部開發人員）根據您的 API 將在回呼的請求主體中發送的資料等正確地實作「外部 API」。

因此，我們接下來要做的是新增程式碼來記錄該「外部 API」應該是什麼樣子，以便從您的 API 接收回呼。

該文件將顯示在您 API 中 /docs 的 Swagger UI 中，它將讓外部開發人員知道如何建置「外部 API」。

此範例沒有實作回呼本身（那可能只是一行程式碼），只有文件部分。

提示

實際的回呼只是一個 HTTP 請求。

當您自己實作回呼時，您可以使用 HTTPX 或 Requests 之類的工具。

編寫回呼函數的文件程式碼

這段程式碼不會在您的應用程式中執行，我們只需要它來*記錄*該*外部 API* 應該是什麼樣子。

但是，您已經知道如何使用 FastAPI 輕鬆地為 API 建立自動文件。

因此，我們將使用相同的知識來記錄*外部 API* 應該是什麼樣子...方法是建立外部 API 應該實作的*路徑操作*（您的 API 將呼叫的操作）。

提示

在編寫程式碼以記錄回呼函數時，想像您是該*外部開發人員*可能會有幫助。並且您目前正在實作*外部 API*，而不是*您的 API*。

暫時採用這種*外部開發人員*的觀點可以幫助您更清楚地了解在哪裡放置參數、主體的 Pydantic 模型、響應等等，以便用於該*外部 API*。

建立回呼函數 APIRouter

首先建立一個新的 APIRouter，其中將包含一個或多個回呼函數。

from typing import Union

from fastapi import APIRouter, FastAPI
from pydantic import BaseModel, HttpUrl

app = FastAPI()


class Invoice(BaseModel):
    id: str
    title: Union[str, None] = None
    customer: str
    total: float


class InvoiceEvent(BaseModel):
    description: str
    paid: bool


class InvoiceEventReceived(BaseModel):
    ok: bool


invoices_callback_router = APIRouter()


@invoices_callback_router.post(
    "{$callback_url}/invoices/{$request.body.id}", response_model=InvoiceEventReceived
)
def invoice_notification(body: InvoiceEvent):
    pass


@app.post("/invoices/", callbacks=invoices_callback_router.routes)
def create_invoice(invoice: Invoice, callback_url: Union[HttpUrl, None] = None):
    """
    Create an invoice.

    This will (let's imagine) let the API user (some external developer) create an
    invoice.

    And this path operation will:

    * Send the invoice to the client.
    * Collect the money from the client.
    * Send a notification back to the API user (the external developer), as a callback.
        * At this point is that the API will somehow send a POST request to the
            external API with the notification of the invoice event
            (e.g. "payment successful").
    """
    # Send the invoice, collect the money, send the notification (the callback)
    return {"msg": "Invoice received"}

建立回呼函數*路徑操作*

要建立回呼函數*路徑操作*，請使用您上面建立的相同 APIRouter。

它看起來應該像普通的 FastAPI *路徑操作*

  • 它應該包含它應該接收的主體的宣告，例如 body: InvoiceEvent。
  • 它還可以包含它應該返回的響應的宣告，例如 response_model=InvoiceEventReceived。
from typing import Union

from fastapi import APIRouter, FastAPI
from pydantic import BaseModel, HttpUrl

app = FastAPI()


class Invoice(BaseModel):
    id: str
    title: Union[str, None] = None
    customer: str
    total: float


class InvoiceEvent(BaseModel):
    description: str
    paid: bool


class InvoiceEventReceived(BaseModel):
    ok: bool


invoices_callback_router = APIRouter()


@invoices_callback_router.post(
    "{$callback_url}/invoices/{$request.body.id}", response_model=InvoiceEventReceived
)
def invoice_notification(body: InvoiceEvent):
    pass


@app.post("/invoices/", callbacks=invoices_callback_router.routes)
def create_invoice(invoice: Invoice, callback_url: Union[HttpUrl, None] = None):
    """
    Create an invoice.

    This will (let's imagine) let the API user (some external developer) create an
    invoice.

    And this path operation will:

    * Send the invoice to the client.
    * Collect the money from the client.
    * Send a notification back to the API user (the external developer), as a callback.
        * At this point is that the API will somehow send a POST request to the
            external API with the notification of the invoice event
            (e.g. "payment successful").
    """
    # Send the invoice, collect the money, send the notification (the callback)
    return {"msg": "Invoice received"}

與正常的*路徑操作*有 2 個主要區別

  • 它不需要有任何實際的程式碼，因為您的應用程式永遠不會呼叫此程式碼。它僅用於記錄*外部 API*。因此，該函數可以只有 pass。
  • *路徑*可以包含 OpenAPI 3 表達式（請參閱下文），其中可以使用帶有參數的變數和發送到*您的 API* 的原始請求的部分內容。

回呼函數路徑表達式

回呼函數*路徑*可以包含一個 OpenAPI 3 表達式，其中可以包含發送到*您的 API* 的原始請求的部分內容。

在這種情況下，它是 str

"{$callback_url}/invoices/{$request.body.id}"

因此，如果您的 API 使用者（外部開發人員）向*您的 API* 發送請求至

https://yourapi.com/invoices/?callback_url=https://www.external.org/events

具有以下 JSON 主體

{
    "id": "2expen51ve",
    "customer": "Mr. Richie Rich",
    "total": "9999"
}

那麼*您的 API* 將處理發票，並在稍後某個時間點，將回呼函數請求發送到 callback_url（*外部 API*）

https://www.external.org/events/invoices/2expen51ve

具有包含如下內容的 JSON 主體

{
    "description": "Payment celebration",
    "paid": true
}

並且它預期來自該*外部 API* 的響應，其 JSON 主體如下

{
    "ok": true
}

提示

請注意，所使用的回呼函數 URL 如何包含在 callback_url 中作為查詢參數接收的 URL（https://www.external.org/events）以及 JSON 主體內部的發票 id（2expen51ve）。

新增回呼函數路由器

此時，您已在上面建立的回呼函數路由器中擁有所需的*回呼函數路徑操作*（*外部開發人員*應該在*外部 API* 中實作的操作）。

現在，使用*您的 API 的路徑操作裝飾器*中的參數 callbacks 來傳遞來自該回呼函數路由器的屬性 .routes（實際上只是路由/*路徑操作*的list）

from typing import Union

from fastapi import APIRouter, FastAPI
from pydantic import BaseModel, HttpUrl

app = FastAPI()


class Invoice(BaseModel):
    id: str
    title: Union[str, None] = None
    customer: str
    total: float


class InvoiceEvent(BaseModel):
    description: str
    paid: bool


class InvoiceEventReceived(BaseModel):
    ok: bool


invoices_callback_router = APIRouter()


@invoices_callback_router.post(
    "{$callback_url}/invoices/{$request.body.id}", response_model=InvoiceEventReceived
)
def invoice_notification(body: InvoiceEvent):
    pass


@app.post("/invoices/", callbacks=invoices_callback_router.routes)
def create_invoice(invoice: Invoice, callback_url: Union[HttpUrl, None] = None):
    """
    Create an invoice.

    This will (let's imagine) let the API user (some external developer) create an
    invoice.

    And this path operation will:

    * Send the invoice to the client.
    * Collect the money from the client.
    * Send a notification back to the API user (the external developer), as a callback.
        * At this point is that the API will somehow send a POST request to the
            external API with the notification of the invoice event
            (e.g. "payment successful").
    """
    # Send the invoice, collect the money, send the notification (the callback)
    return {"msg": "Invoice received"}

提示

請注意，您傳遞給 callback= 的不是路由器本身 (invoices_callback_router)，而是它的屬性 .routes，如 invoices_callback_router.routes。

查看文件

現在您可以啟動您的應用程式並前往 http://127.0.0.1:8000/docs。

您將會看到您的文件，其中包含一個針對您的*路徑操作*的「回呼」區段，顯示*外部 API* 應該呈現的樣子。