Saltar a contenido

Callbacks de OpenAPI

Podr铆as crear una API con una path operation que podr铆a desencadenar un request a una API externa creada por alguien m谩s (probablemente el mismo desarrollador que estar铆a usando tu API).

El proceso que ocurre cuando tu aplicaci贸n API llama a la API externa se llama un "callback". Porque el software que escribi贸 el desarrollador externo env铆a un request a tu API y luego tu API hace un callback, enviando un request a una API externa (que probablemente fue creada por el mismo desarrollador).

En este caso, podr铆as querer documentar c贸mo esa API externa deber铆a verse. Qu茅 path operation deber铆a tener, qu茅 cuerpo deber铆a esperar, qu茅 response deber铆a devolver, etc.

Una aplicaci贸n con callbacks

Veamos todo esto con un ejemplo.

Imagina que desarrollas una aplicaci贸n que permite crear facturas.

Estas facturas tendr谩n un id, title (opcional), customer y total.

El usuario de tu API (un desarrollador externo) crear谩 una factura en tu API con un request POST.

Luego tu API (imaginemos):

  • Enviar谩 la factura a alg煤n cliente del desarrollador externo.
  • Recoger谩 el dinero.
  • Enviar谩 una notificaci贸n de vuelta al usuario de la API (el desarrollador externo).
    • Esto se har谩 enviando un request POST (desde tu API) a alguna API externa proporcionada por ese desarrollador externo (este es el "callback").

La aplicaci贸n normal de FastAPI

Primero veamos c贸mo se ver铆a la aplicaci贸n API normal antes de agregar el callback.

Tendr谩 una path operation que recibir谩 un cuerpo Invoice, y un par谩metro de query callback_url que contendr谩 la URL para el callback.

Esta parte es bastante normal, probablemente ya est茅s familiarizado con la mayor parte del c贸digo:

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

app = FastAPI()


class Invoice(BaseModel):
    id: str
    title: 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: 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"}
馃 Other versions and variants
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"}

Consejo

El par谩metro de query callback_url utiliza un tipo Url de Pydantic.

Lo 煤nico nuevo es callbacks=invoices_callback_router.routes como un argumento para el decorador de path operation. Veremos qu茅 es eso a continuaci贸n.

Documentar el callback

El c贸digo real del callback depender谩 mucho de tu propia aplicaci贸n API.

Y probablemente variar谩 mucho de una aplicaci贸n a otra.

Podr铆a ser solo una o dos l铆neas de c贸digo, como:

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

Pero posiblemente la parte m谩s importante del callback es asegurarse de que el usuario de tu API (el desarrollador externo) implemente la API externa correctamente, de acuerdo con los datos que tu API va a enviar en el request body del callback, etc.

As铆 que, lo que haremos a continuaci贸n es agregar el c贸digo para documentar c贸mo deber铆a verse esa API externa para recibir el callback de tu API.

Esa documentaci贸n aparecer谩 en la Swagger UI en /docs en tu API, y permitir谩 a los desarrolladores externos saber c贸mo construir la API externa.

Este ejemplo no implementa el callback en s铆 (eso podr铆a ser solo una l铆nea de c贸digo), solo la parte de documentaci贸n.

Consejo

El callback real es solo un request HTTP.

Cuando implementes el callback t煤 mismo, podr铆as usar algo como HTTPX o Requests.

Escribe el c贸digo de documentaci贸n del callback

Este c贸digo no se ejecutar谩 en tu aplicaci贸n, solo lo necesitamos para documentar c贸mo deber铆a verse esa API externa.

Pero ya sabes c贸mo crear f谩cilmente documentaci贸n autom谩tica para una API con FastAPI.

As铆 que vamos a usar ese mismo conocimiento para documentar c贸mo deber铆a verse la API externa... creando la(s) path operation(s) que la API externa deber铆a implementar (las que tu API va a llamar).

Consejo

Cuando escribas el c贸digo para documentar un callback, podr铆a ser 煤til imaginar que eres ese desarrollador externo. Y que actualmente est谩s implementando la API externa, no tu API.

Adoptar temporalmente este punto de vista (del desarrollador externo) puede ayudarte a sentir que es m谩s obvio d贸nde poner los par谩metros, el modelo de Pydantic para el body, para el response, etc. para esa API externa.

Crea un APIRouter de callback

Primero crea un nuevo APIRouter que contendr谩 uno o m谩s callbacks.

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

app = FastAPI()


class Invoice(BaseModel):
    id: str
    title: 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: 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"}
馃 Other versions and variants
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"}

Crea la path operation del callback

Para crear la path operation del callback usa el mismo APIRouter que creaste arriba.

Deber铆a verse como una path operation normal de FastAPI:

  • Probablemente deber铆a tener una declaraci贸n del body que deber铆a recibir, por ejemplo body: InvoiceEvent.
  • Y tambi茅n podr铆a tener una declaraci贸n del response que deber铆a devolver, por ejemplo response_model=InvoiceEventReceived.
from fastapi import APIRouter, FastAPI
from pydantic import BaseModel, HttpUrl

app = FastAPI()


class Invoice(BaseModel):
    id: str
    title: 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: 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"}
馃 Other versions and variants
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"}

Hay 2 diferencias principales respecto a una path operation normal:

  • No necesita tener ning煤n c贸digo real, porque tu aplicaci贸n nunca llamar谩 a este c贸digo. Solo se usa para documentar la API externa. As铆 que, la funci贸n podr铆a simplemente tener pass.
  • El path puede contener una expresi贸n OpenAPI 3 (ver m谩s abajo) donde puede usar variables con par谩metros y partes del request original enviado a tu API.

La expresi贸n del path del callback

El path del callback puede tener una expresi贸n OpenAPI 3 que puede contener partes del request original enviado a tu API.

En este caso, es el str:

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

Entonces, si el usuario de tu API (el desarrollador externo) env铆a un request a tu API a:

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

con un JSON body de:

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

luego tu API procesar谩 la factura y, en alg煤n momento despu茅s, enviar谩 un request de callback al callback_url (la API externa):

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

con un JSON body que contiene algo como:

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

y esperar铆a un response de esa API externa con un JSON body como:

{
    "ok": true
}

Consejo

Observa c贸mo la URL del callback utilizada contiene la URL recibida como par谩metro de query en callback_url (https://www.external.org/events) y tambi茅n el id de la factura desde dentro del JSON body (2expen51ve).

Agrega el router de callback

En este punto tienes las path operation(s) del callback necesarias (las que el desarrollador externo deber铆a implementar en la API externa) en el router de callback que creaste arriba.

Ahora usa el par谩metro callbacks en el decorador de path operation de tu API para pasar el atributo .routes (que en realidad es solo un list de rutas/path operations) de ese router de callback:

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

app = FastAPI()


class Invoice(BaseModel):
    id: str
    title: 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: 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"}
馃 Other versions and variants
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"}

Consejo

Observa que no est谩s pasando el router en s铆 (invoices_callback_router) a callback=, sino el atributo .routes, como en invoices_callback_router.routes.

Revisa la documentaci贸n

Ahora puedes iniciar tu aplicaci贸n e ir a http://127.0.0.1:8000/docs.

Ver谩s tu documentaci贸n incluyendo una secci贸n de "Callbacks" para tu path operation que muestra c贸mo deber铆a verse la API externa: