Callbacks de OpenAPI¶
Podrías crear una API con una path operation que podría disparar una 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 app de API llama a la API externa se llama "callback". Porque el software que el desarrollador externo escribió envía una request a tu API y luego tu API llama de vuelta, enviando una request a una API externa (que probablemente fue creada por el mismo desarrollador).
En este caso, podrías querer documentar cómo debería verse esa API externa. Qué path operation debería tener, qué body debería esperar, qué response debería retornar, etc.
Una app con callbacks¶
Veamos todo esto con un ejemplo.
Imagina que desarrollas una app que permite crear facturas.
Estas facturas tendrán un id, un title (opcional), un customer, y un total.
El usuario de tu API (un desarrollador externo) creará una factura en tu API con una request POST.
Luego tu API (imaginemos):
- Enviar la factura a algún cliente del desarrollador externo.
- Cobrar el dinero.
- Send a notification back to the API user (the external developer).
- Esto se hará enviando una request POST (desde tu API) a alguna API externa proporcionada por ese desarrollador externo (este es el "callback").
La app normal de FastAPI¶
Veamos primero cómo se vería la app normal de la API antes de añadir el callback.
Tendrá una path operation que recibirá un body Invoice, y un query parameter callback_url que contendrá la URL para el callback.
Esta parte es bastante normal, la mayoría del código probablemente ya te sea familiar:
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"}
Consejo
El query parameter callback_url usa un tipo Url de Pydantic.
Lo único nuevo es callbacks=invoices_callback_router.routes como argumento del decorador de la path operation. Veremos qué es eso a continuación.
Documentar el callback¶
El código real del callback dependerá mucho de tu propia app de API.
Y probablemente variará mucho de una app a la siguiente.
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 asegurar que el usuario de tu API (el desarrollador externo) implemente la API externa correctamente, según los datos que tu API va a enviar en el body de la request del callback, etc.
Así que, lo que haremos a continuación es añadir 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 el Swagger UI en /docs de tu API, y les informará a los desarrolladores externos cómo construir la API externa.
Este ejemplo no implementa el callback en sí (que podría ser solo una línea de código), solo la parte de documentación.
Consejo
El callback real es solo una request HTTP.
Al implementar el callback tú mismo, podrías usar algo como HTTPX o Requests.
Escribir el código de documentación del callback¶
Este código no se ejecutará en tu app, 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 las path operation(s) que la API externa debería implementar (las que tu API llamará).
Consejo
Al escribir el código para documentar un callback, puede ser útil imaginar que eres ese desarrollador externo. Y que estás actualmente 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 la response, etc. de esa API externa.
Crear un APIRouter para el 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"}
Crear la path operation del callback¶
Para crear la path operation del callback usa el mismo APIRouter que creaste arriba.
Debería verse igual que 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 de la response que debería retornar, 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"}
Hay 2 diferencias principales con una path operation normal:
- No necesita tener ningún código real, porque tu app nunca llamará 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 de OpenAPI 3 (ver más abajo) donde puede usar variables con parámetros y partes de la request original enviada a tu API.
La expresión path del callback¶
El path del callback puede tener una expresión de OpenAPI 3 que puede contener partes de la request original enviada a tu API.
En este caso, es el str:
"{$callback_url}/invoices/{$request.body.id}"
Así que, si el usuario de tu API (el desarrollador externo) envía una request a tu API a:
https://yourapi.com/invoices/?callback_url=https://www.external.org/events
con un body JSON de:
{
"id": "2expen51ve",
"customer": "Mr. Richie Rich",
"total": "9999"
}
entonces tu API procesará la factura, y en algún momento después, enviará una request de callback al callback_url (la API externa):
https://www.external.org/events/invoices/2expen51ve
con un body JSON conteniendo algo como:
{
"description": "Payment celebration",
"paid": true
}
y esperaría una response de esa API externa con un body JSON como:
{
"ok": true
}
Consejo
Notá cómo la URL del callback usada contiene la URL recibida como query parameter en callback_url (https://www.external.org/events) y también el id de la factura desde dentro del body JSON (2expen51ve).
Añadir el router del callback¶
En este punto tienes las callback path operation(s) 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 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"}
Consejo
Notá que no estás pasando el router mismo (invoices_callback_router) a callbacks=, sino sus .routes, como en invoices_callback_router.routes. FastAPI usará esas routes para generar la documentación de OpenAPI del callback.
Revisar la documentación¶
Ahora puedes iniciar tu app e ir a http://127.0.0.1:8000/docs.
Verás tu documentación incluyendo una sección "Callbacks" para tu path operation que muestra cómo debería verse la API externa:
