FastAPI is a modern, high-performance web framework for building APIs with Python. One of its advanced features is the ability to implement OpenAPI callbacks for asynchronous communication with external services. This can greatly improve the efficiency and responsiveness of your API, especially when dealing with long-running operations or integrating with third-party services.

To get started with OpenAPI callbacks in FastAPI, you’ll first need to ensure you have FastAPI and its dependencies installed. You can do this using pip:

pip install fastapi[all]

Now, let’s dive into implementing callbacks. The basic idea is that your API will initiate a request to an external service, and that service will later call back to your API with the results. This is particularly useful for operations that might take a while to complete.

Here’s a simple example of how you might structure a FastAPI application with a callback:

from fastapi import FastAPI, BackgroundTasks
from pydantic import BaseModel, HttpUrl
import httpx
import asyncio

app = FastAPI()

class CallbackModel(BaseModel):
    url: HttpUrl

class Item(BaseModel):
    id: str
    status: str = "processing"

@app.post("/process-item/{item_id}")
async def process_item(item_id: str, callback: CallbackModel, background_tasks: BackgroundTasks):
    item = Item(id=item_id)
    background_tasks.add_task(process_and_callback, item, callback.url)
    return {"message": "Processing started", "item": item}

async def process_and_callback(item: Item, callback_url: HttpUrl):
    # Simulate some processing time
    await asyncio.sleep(10)
    item.status = "completed"
    
    async with httpx.AsyncClient() as client:
        await client.post(str(callback_url), json=item.dict())

@app.post("/callback")
async def callback(item: Item):
    print(f"Received callback for item {item.id}: {item.status}")
    return {"message": "Callback received"}

In this example, we define a /process-item/{item_id} endpoint that accepts a callback URL. It immediately returns a response to the client, but starts a background task to process the item. Once the processing is complete, it sends a POST request to the provided callback URL with the updated item status.

The /callback endpoint is where the external service would send its response. In a real-world scenario, you’d probably want to update your database or perform some other action here.

One of the cool things about FastAPI is that it automatically generates OpenAPI (Swagger) documentation for your API. This includes information about the callbacks, making it easier for other developers to understand and integrate with your API.

To make your callbacks more robust, you might want to implement retry logic in case the callback fails. Here’s how you could modify the process_and_callback function to include retries:

from tenacity import retry, stop_after_attempt, wait_exponential

@retry(stop=stop_after_attempt(3), wait=wait_exponential(multiplier=1, min=4, max=10))
async def send_callback(client: httpx.AsyncClient, url: str, data: dict):
    response = await client.post(url, json=data)
    response.raise_for_status()

async def process_and_callback(item: Item, callback_url: HttpUrl):
    await asyncio.sleep(10)
    item.status = "completed"
    
    async with httpx.AsyncClient() as client:
        try:
            await send_callback(client, str(callback_url), item.dict())
        except Exception as e:
            print(f"Failed to send callback after 3 attempts: {e}")

This version will retry the callback up to 3 times, with exponential backoff between attempts.

When working with callbacks, it’s important to consider security. You don’t want just anyone to be able to send callbacks to your API. One way to handle this is by including a secret token in the callback URL:

from fastapi import FastAPI, BackgroundTasks, HTTPException, Depends
from fastapi.security import APIKeyHeader
import secrets

app = FastAPI()

API_KEY_NAME = "X-API-Key"
api_key_header = APIKeyHeader(name=API_KEY_NAME, auto_error=False)

def verify_api_key(api_key: str = Depends(api_key_header)):
    if api_key != "your-secret-api-key":
        raise HTTPException(status_code=403, detail="Invalid API Key")

@app.post("/process-item/{item_id}")
async def process_item(item_id: str, callback: CallbackModel, background_tasks: BackgroundTasks):
    callback_token = secrets.token_urlsafe()
    callback_url = f"{callback.url}?token={callback_token}"
    item = Item(id=item_id)
    background_tasks.add_task(process_and_callback, item, callback_url)
    return {"message": "Processing started", "item": item, "callback_token": callback_token}

@app.post("/callback")
async def callback(item: Item, token: str, api_key: str = Depends(verify_api_key)):
    # Verify the token here
    print(f"Received callback for item {item.id}: {item.status}")
    return {"message": "Callback received"}

In this version, we generate a unique token for each callback and include it in the callback URL. We also add API key verification to the callback endpoint for an extra layer of security.

Callbacks are just one way to handle asynchronous operations in FastAPI. Another approach is to use WebSockets for real-time communication. Here’s a simple example of how you might implement a WebSocket endpoint:

from fastapi import FastAPI, WebSocket
from fastapi.websockets import WebSocketDisconnect

app = FastAPI()

@app.websocket("/ws")
async def websocket_endpoint(websocket: WebSocket):
    await websocket.accept()
    try:
        while True:
            data = await websocket.receive_text()
            await websocket.send_text(f"Message text was: {data}")
    except WebSocketDisconnect:
        print("Client disconnected")

This creates a WebSocket endpoint that echoes back any messages it receives. You could modify this to send updates about long-running processes instead of using callbacks.

When working with external services, it’s often a good idea to implement circuit breakers to prevent cascading failures. The circuitbreaker library works well with FastAPI:

from circuitbreaker import circuit

@circuit(failure_threshold=5, recovery_timeout=30)
async def call_external_service(client: httpx.AsyncClient, url: str):
    response = await client.get(url)
    response.raise_for_status()
    return response.json()

@app.get("/external-data")
async def get_external_data():
    async with httpx.AsyncClient() as client:
        try:
            data = await call_external_service(client, "https://api.example.com/data")
            return {"data": data}
        except Exception as e:
            return {"error": str(e)}

This will automatically “open” the circuit after 5 failures, preventing further calls to the external service for 30 seconds.

As your FastAPI application grows, you might want to consider structuring it into multiple files. Here’s an example of how you might organize a larger application:

myapp/
    __init__.py
    main.py
    api/
        __init__.py
        items.py
        users.py
    models/
        __init__.py
        item.py
        user.py
    services/
        __init__.py
        item_service.py
        user_service.py
    config.py
    dependencies.py

In this structure, main.py would be your entry point:

from fastapi import FastAPI
from .api import items, users

app = FastAPI()

app.include_router(items.router)
app.include_router(users.router)

And api/items.py might look like:

from fastapi import APIRouter, Depends
from ..models.item import Item
from ..services.item_service import ItemService
from ..dependencies import get_item_service

router = APIRouter()

@router.post("/items")
async def create_item(item: Item, service: ItemService = Depends(get_item_service)):
    return await service.create_item(item)

This structure helps keep your code organized and maintainable as your application grows.

When it comes to testing your FastAPI application, the TestClient class makes it easy:

from fastapi.testclient import TestClient
from .main import app

client = TestClient(app)

def test_read_main():
    response = client.get("/")
    assert response.status_code == 200
    assert response.json() == {"message": "Hello World"}

def test_create_item():
    response = client.post(
        "/items/",
        json={"id": "foobar", "title": "Foo Bar", "description": "The Foo Barters"},
    )
    assert response.status_code == 200
    assert response.json() == {
        "id": "foobar",
        "title": "Foo Bar",
        "description": "The Foo Barters",
    }

These tests can be run using pytest, making it easy to ensure your API behaves correctly as you make changes.

As you can see, FastAPI provides a powerful set of tools for building advanced APIs. From OpenAPI callbacks to WebSockets, from circuit breakers to comprehensive testing, it offers everything you need to create robust, efficient, and scalable web services. The key is to start simple and gradually incorporate these advanced features as your needs grow. Happy coding!