How to implement callback functionality in FastAPI?

Question

I'm trying to implement a service that will get a request from an external API, do some work (which might take time) and then return a response to the external API with the parsed data. However I'm at a loss on how to achieve this. I'm using FastAPI as my API service and have been looking at the following documentation: OpenAPI Callbacks By following that documentation I can get the OpenAPI docs looking all pretty and nice. However I'm stumped on how to implement the actual callback and the docs don't have much information about that. My current implementation is as follows:

from typing import Union

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

import requests
import time

from threading import Thread

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)
async def create_invoice(invoice: Invoice, callback_url: Union[AnyHttpUrl, None] = None):
    # Send the invoice, collect the money, send the notification (the callback)
    thread = Thread(target=do_invoice(invoice, callback_url))
    thread.start()
    return {"msg": "Invoice received"}

def do_invoice(invoice: Invoice, callback_url: AnyHttpUrl):
    time.sleep(10)
    url = callback_url + "/invoices/" + invoice.id
    json = {
        "data": ["Payment celebration"],
    }
    requests.post(url=url, json=json)

I thought putting the actual callback in a separate thread might work and that the {"msg": "Invoice received"} would be returned immediately and then 10s later the external api would recieve the result from the do_invoice function. But this doesn't seem to be the case so perhaps I'm doing something wrong.

I've also tried putting the logic in the invoice_notification function but that doesn't seem to do anything at all.

So what is the correct to implement a callback like the one I want? Thankful for any help!

Chris · Accepted Answer · 2023-02-17T06:57:22.547

I thought putting the actual callback in a separate thread might work and that the {"msg": "Invoice received"} would be returned immediately and then 10s later the external api would recieve the result from the do_invoice function. But this doesn't seem to be the case so perhaps I'm doing something wrong.

If you would like to run a task after the response has been sent, you could use a BackgroundTask, as demonstrated in this answer, as well as here and here. If you instead would like to wait for the task to finish before returning the response, you could run the task in either an external ThreadPool or ProcessPool (depending on the nature of the task) and await it, as explained in this detailed answer.

I would also strongly recommend using the httpx library in an async environment such as FastAPI, instead of using Python requests—you may find details and working examples here, as well as here and here.

How to implement callback functionality in FastAPI?

1 Answers1