Event-driven architecture in FastAPI

Introduction

Event-driven architecture helps your app react to actions or changes instantly. It makes your app more flexible and easier to manage by focusing on events.

You want your FastAPI app to respond immediately when a user uploads a file.

You need to trigger background tasks after a user submits a form.

Your app should update other parts automatically when data changes.

You want to decouple parts of your app so they work independently.

You want to handle notifications or logs when certain actions happen.

Syntax

FastAPI

from fastapi import FastAPI
from fastapi import BackgroundTasks

app = FastAPI()

@app.post("/items/")
async def create_item(item: dict, background_tasks: BackgroundTasks):
    background_tasks.add_task(process_item, item)
    return {"message": "Item received"}

def process_item(item: dict):
    # handle the event here
    print(f"Processing item: {item}")

Use BackgroundTasks to run functions after the main request finishes.

Define event handlers as normal functions and add them to background tasks.

Examples

This example sends a welcome email after the API call finishes.

FastAPI

from fastapi import FastAPI, BackgroundTasks

app = FastAPI()

@app.post("/send-email/")
async def send_email(background_tasks: BackgroundTasks):
    background_tasks.add_task(send_welcome_email)
    return {"message": "Email will be sent"}

def send_welcome_email():
    print("Sending welcome email...")

This example logs user actions to a file without delaying the response.

FastAPI

from fastapi import FastAPI, BackgroundTasks

app = FastAPI()

@app.post("/log-action/")
async def log_action(action: str, background_tasks: BackgroundTasks):
    background_tasks.add_task(log_to_file, action)
    return {"message": "Action logged"}

def log_to_file(action: str):
    with open("actions.log", "a") as f:
        f.write(f"Action: {action}\n")

Sample Program

This FastAPI app accepts data and immediately returns a response. Meanwhile, it processes the data in the background as an event.

FastAPI

from fastapi import FastAPI, BackgroundTasks
from fastapi.responses import JSONResponse

app = FastAPI()

@app.post("/process-data/")
async def process_data(data: dict, background_tasks: BackgroundTasks):
    background_tasks.add_task(handle_event, data)
    return JSONResponse(content={"status": "Data received"})

def handle_event(data: dict):
    print(f"Event received with data: {data}")
    # Imagine processing or notifying other services here

Important Notes

Background tasks run after the response is sent, so users don't wait.

Use event-driven design to keep your app responsive and scalable.

FastAPI's BackgroundTasks is a simple way to start with event-driven patterns.

Summary

Event-driven architecture lets your app react to actions as they happen.

FastAPI uses BackgroundTasks to handle events without blocking users.

This approach makes your app faster and easier to maintain.