コンテンツへスキップ

OpenAPI コールバック

あなたは、別の誰かが作成した(おそらくあなたのAPIを*使用*する同じ開発者)外部APIへのリクエストをトリガーできるパスオペレーションを持つAPIを作成できます。

あなたのAPIアプリが外部APIを呼び出す際に発生するプロセスは、「コールバック」と呼ばれます。外部開発者が書いたソフトウェアがあなたのAPIにリクエストを送信し、次にあなたのAPIがコールバックして、外部API(おそらく同じ開発者によって作成された)にリクエストを送信するためです。

この場合、その外部APIがどのように見えるべきかを文書化したい場合があります。どのようなパスオペレーションを持つべきか、どのようなボディを期待すべきか、どのようなレスポンスを返すべきかなどです。

コールバックを含むアプリ

例を通してこれを見てみましょう。

請求書の作成を可能にするアプリを開発すると想像してください。

これらの請求書には、`id`、`title`(オプション)、`customer`、`total`があります。

あなたのAPIのユーザー(外部開発者)は、POSTリクエストであなたのAPIに請求書を作成します。

その後、あなたのAPIは(想像してください)

  • 請求書を外部開発者の顧客に送信します。
  • お金を集めます。
  • APIユーザー(外部開発者)に通知を送信します。
    • これは、(あなたのAPIから)その外部開発者が提供する外部APIへのPOSTリクエストを送信することによって行われます(これが「コールバック」です)。

通常の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アプリに大きく依存します。

そして、アプリごとに大きく異なる可能性があります。

これは、次の1〜2行のコードのようなものでしょう。

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を構築する方法を知ることができます。

この例では、コールバック自体(それは1行のコードかもしれません)は実装せず、ドキュメントの部分のみを実装します。

ヒント

実際のコールバックは、単なるHTTPリクエストです。

自分でコールバックを実装する際には、HTTPXRequestsなどを使用できます。

コールバックドキュメントコードの記述

このコードはあなたのアプリケーションでは実行されません。これは、外部APIがどのようなものになるべきかを記述するためにのみ必要です。

しかし、FastAPIを使ってAPIの自動ドキュメントを簡単に作成する方法を既に知っているでしょう。

そのため、同じ知識を使って外部APIがどのようなものになるべきかを記述します…外部APIが実装する必要があるパス操作(あなたのAPIが呼び出すもの)を作成することによって。

ヒント

コールバックを記述するコードを作成する際には、あなたがその外部開発者であると想像することが役立つかもしれません。そして、あなたは現在あなたのAPIではなく、外部APIを実装しているところです。

この視点(外部開発者の視点)を一時的に採用することで、その外部APIのパラメータ、ボディのPydanticモデル、レスポンスなどをどこに配置すればよいかがより明確に感じられるようになるでしょう。

コールバックAPIRouterの作成

まず、1つ以上のコールバックを含む新しい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に送信された元の要求のパラメータと部分を使用する変数を使用できます。

コールバックパス式の表現

コールバックパスには、あなたのAPIに送信された元の要求の一部を含めることができるOpenAPI 3式を含めることができます。

この場合、それは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_urlhttps://www.external.org/events)のクエリパラメータとして受信したURLと、JSONボディ内(2expen51ve)の請求書idの両方を含んでいることに注意してください。

コールバックルーターの追加

この時点で、(外部開発者外部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がどのように見えるべきかを示すパス操作の「コールバック」セクションを含むドキュメントが表示されます。