Event-driven architecture in FastAPI - Mini Project: Build & Apply

Building a Simple Event-Driven API with FastAPI

📖 Scenario: You are creating a small web API that handles user registrations. Instead of processing everything in one step, you want to use an event-driven approach. When a user registers, an event is triggered to send a welcome email asynchronously.

🎯 Goal: Build a FastAPI application that stores user data, triggers a registration event, and handles that event to simulate sending a welcome email.

📋 What You'll Learn

Create a list called users to store user dictionaries with username and email keys.

Create a variable called event_queue as a list to hold event messages.

Write a function register_user that adds a user dictionary to users and appends a registration event message to event_queue.

Add a FastAPI route /register that accepts JSON with username and email, calls register_user, and returns a confirmation message.

💡 Why This Matters

🌍 Real World

Event-driven architecture helps build scalable and responsive web applications by decoupling user actions from background processing.

💼 Career

Understanding event-driven design and FastAPI is valuable for backend developers building modern APIs and microservices.

1

Create the initial data structures

Create a list called users that is empty and a list called event_queue that is also empty.

FastAPI

# Create empty lists for users and event_queue
# Your code here

Need a hint?

Use empty square brackets [] to create empty lists.

2

Create the user registration function

Define a function called register_user that takes username and email as parameters. Inside the function, create a dictionary with keys 'username' and 'email' and add it to the users list. Then append a string f"User {username} registered" to the event_queue list.

FastAPI

users = []
event_queue = []

# Define register_user function here
# Your code here

Need a hint?

Use def to define the function and append() to add items to lists.

3

Create the FastAPI app and registration route

Import FastAPI from fastapi. Create an instance called app. Add a POST route /register that accepts JSON with username and email. Inside the route function, call register_user with the received data and return a dictionary with key message and value "User registered successfully".

FastAPI

users = []
event_queue = []

def register_user(username, email):
    user = {'username': username, 'email': email}
    users.append(user)
    event_queue.append(f"User {username} registered")

# Import FastAPI and create app instance
# Add POST /register route
# Your code here

Need a hint?

Use @app.post('/register') decorator and define an async function that accepts a Pydantic model.

4

Add event processing function

Define a function called process_events that loops over the event_queue list. For each event, remove it from the list and simulate sending an email by creating a string f"Sending email for event: {event}". Call process_events at the end of the register route function.

FastAPI

from fastapi import FastAPI
from pydantic import BaseModel

users = []
event_queue = []

def register_user(username, email):
    user = {'username': username, 'email': email}
    users.append(user)
    event_queue.append(f"User {username} registered")

app = FastAPI()

class User(BaseModel):
    username: str
    email: str

@app.post('/register')
async def register(user: User):
    register_user(user.username, user.email)
    # Call process_events here
    # Your code here

def process_events():
    # Your code here

Need a hint?

Use a while loop to process all events and pop(0) to remove the first event.