[Python] How to implement callback functionality in FastAPI?

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!

Continue reading...