🍭Appendix B

Event-Driven Communication in FastAPI

Event-driven communication in FastAPI, or in web development contexts generally, refers to a programming pattern where the flow of the program is determined by events or changes in state. This is a particularly useful paradigm in asynchronous programming and is often used in web applications to handle tasks such as sending notifications, processing data in the background, and integrating real-time features.

FastAPI, being an asynchronous framework, leverages Python's asyncio library to enable event-driven communication. This allows FastAPI applications to perform non-blocking network and I/O operations efficiently.

---

WebSocket Communication

One of the most direct forms of event-driven communication in FastAPI is through WebSockets. WebSockets allow for two-way interactive communication sessions between the user's browser and a server. FastAPI supports WebSocket routes that can listen for events from the client (such as messages sent from a web page) and send responses back in real-time.

from fastapi import FastAPI, WebSocket

app = FastAPI()

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

Background Tasks

FastAPI allows you to define background tasks that are executed after returning a response to the client. This is useful for event-driven actions that need to happen after a request, but don't necessarily need to delay the response. For example, sending an email confirmation after a user registration.

from fastapi import FastAPI, BackgroundTasks

app = FastAPI()

def write_log(message: str):
    # Imagine this writes to a log file
    pass

@app.post("/send-notification/")
async def send_notification(email: str, background_tasks: BackgroundTasks):
    background_tasks.add_task(write_log, f"Notification sent to {email}")
    return {"message": "Notification sent in the background"}

Starlette Events

Since FastAPI is built on top of Starlette, you can also use Starlette's event hooks for startup and shutdown events. These can be used to set up and tear down resources (like database connections) when your application starts or stops.

from fastapi import FastAPI

app = FastAPI()

@app.on_event("startup")
async def startup_event():
    # Perform setup actions here, e.g., connect to a database
    pass

@app.on_event("shutdown")
async def shutdown_event():
    # Perform teardown actions here, e.g., close database connections
    pass

---

Summary

Event-driven communication in FastAPI leverages asynchronous programming to efficiently handle real-time data, background processing, and the life cycle of the application. By integrating WebSockets, background tasks, and application event hooks, FastAPI provides a robust set of tools for building scalable, real-time web applications.