Event-driven architecture in FastAPI - Practice Problems & Coding Challenges

Challenge - 5 Problems

❓ component_behavior

intermediate

What is the output when an event is emitted and handled in FastAPI?

Consider this FastAPI code snippet using an event system:

from fastapi import FastAPI
app = FastAPI()

subscribers = []

def on_event(data):
    subscribers.append(data)

@app.post('/emit')
async def emit_event(payload: dict):
    on_event(payload['message'])
    return {'status': 'event received'}

@app.get('/subscribers')
async def get_subscribers():
    return {'messages': subscribers}

What will be the output of a GET request to /subscribers after posting {"message": "hello"} to /emit once?

FastAPI

from fastapi import FastAPI
app = FastAPI()

subscribers = []

def on_event(data):
    subscribers.append(data)

@app.post('/emit')
async def emit_event(payload: dict):
    on_event(payload['message'])
    return {'status': 'event received'}

@app.get('/subscribers')
async def get_subscribers():
    return {'messages': subscribers}

A {"messages": ["hello"]}

B {"messages": []}

C {"messages": ["{'message': 'hello'}"]}

D {"messages": [null]}

💡 Hint

Think about what the on_event function does with the data it receives.

The on_event function appends the message string to the subscribers list. Posting to /emit triggers this function with the message "hello". So, the /subscribers endpoint returns the list containing "hello".

❓ lifecycle

intermediate

When does FastAPI's startup event run in an event-driven app?

In FastAPI, you can register a function to run on startup using @app.on_event('startup'). When exactly does this function run?

FastAPI

from fastapi import FastAPI
app = FastAPI()

@app.on_event('startup')
async def startup_event():
    print('App is starting')

A Only when the first request is received

B Every time a client sends a request

C When the FastAPI server starts before handling any requests

D When the server shuts down

💡 Hint

Think about the server lifecycle phases: startup, running, shutdown.

The startup event runs once when the FastAPI server starts, before it handles any requests. It is used to prepare resources or initialize state.

🔧 Debug

advanced

Why does this FastAPI event handler not update the shared state?

Look at this FastAPI code:

from fastapi import FastAPI
app = FastAPI()

state = {'count': 0}

def increment():
    state = {'count': state['count'] + 1}

@app.post('/inc')
async def increment_endpoint():
    increment()
    return state

Why does calling /inc never increase count?

FastAPI

from fastapi import FastAPI
app = FastAPI()

state = {'count': 0}

def increment():
    state = {'count': state['count'] + 1}

@app.post('/inc')
async def increment_endpoint():
    increment()
    return state

A FastAPI does not allow modifying global variables inside functions

B The increment function creates a new local variable 'state' instead of modifying the global one

C The state dictionary is immutable and cannot be changed

D The increment function is never called

💡 Hint

Check variable scope and assignment inside the increment function.

Inside increment, assigning to state creates a new local variable, shadowing the global state. The global state remains unchanged, so the count never increases.

📝 Syntax

advanced

Which FastAPI event handler code will raise a SyntaxError?

Identify the option that contains a syntax error in defining an event handler in FastAPI.

A

@app.on_event('startup')
async def startup_event()
    print('Starting up')

B

@app.on_event('startup')
async def startup_event():
    print('Starting up')

C

@app.on_event('shutdown')
async def shutdown_event():
    print('Shutting down')

D

@app.on_event('shutdown')
def shutdown_event():
    print('Shutting down')

💡 Hint

Look carefully at function definitions and punctuation.

Option A is missing a colon at the end of the function definition line, causing a SyntaxError.

🧠 Conceptual

expert

What is a key benefit of event-driven architecture in FastAPI apps?

Choose the best explanation of why event-driven architecture is useful in FastAPI applications.

A It forces all code to run synchronously, simplifying debugging

B It requires fewer lines of code by removing all function definitions

C It eliminates the need for any shared state or data storage

D It allows decoupling components so they react to events independently, improving scalability and maintainability

💡 Hint

Think about how events help parts of an app communicate without tight connections.

Event-driven architecture lets components respond to events independently, which helps scale the app and makes it easier to change parts without breaking others.

Practice

1. What is the main benefit of using event-driven architecture in FastAPI?

easy

A. It allows the app to react to actions immediately without waiting.

B. It makes the app run only on a single thread.

C. It requires no code to handle user requests.

D. It disables background tasks for faster response.

Solution

1. Step 1: Understand event-driven architecture

   Event-driven architecture lets the app respond to events as they happen, improving responsiveness.

2. Step 2: Connect to FastAPI's use of BackgroundTasks

   FastAPI uses BackgroundTasks to handle events without blocking the main user request, enabling immediate reactions.

3. Final Answer:

   It allows the app to react to actions immediately without waiting. -> Option A

4. Quick Check:

   Event-driven = Immediate reaction [OK]

Hint: Event-driven means reacting instantly to actions [OK]

Common Mistakes:

• Thinking event-driven means single-threaded only
• Believing no code is needed for requests
• Assuming background tasks are disabled

2. Which of the following is the correct way to import BackgroundTasks in FastAPI?

easy

A. import BackgroundTasks from fastapi

B. from fastapi.background import Tasks

C. from fastapi import BackgroundTasks

D. import fastapi.BackgroundTasks

Solution

1. Step 1: Recall FastAPI import syntax

   FastAPI uses standard Python import syntax: from module import class/function.

2. Step 2: Identify correct module and class

   The BackgroundTasks class is imported from fastapi directly, so the correct syntax is 'from fastapi import BackgroundTasks'.

3. Final Answer:

   from fastapi import BackgroundTasks -> Option C

4. Quick Check:

   Correct import syntax = from fastapi import BackgroundTasks [OK]

Hint: Use 'from fastapi import BackgroundTasks' to import correctly [OK]

Common Mistakes:

• Using wrong import order or syntax
• Trying to import from a non-existent submodule
• Using dot notation incorrectly in import

3. What will be the output when this FastAPI endpoint is called?

from fastapi import FastAPI, BackgroundTasks
app = FastAPI()

results = []

def write_log(message: str):
    results.append(message)

@app.post('/send')
async def send_message(background_tasks: BackgroundTasks):
    background_tasks.add_task(write_log, 'Message sent')
    return {'status': 'ok'}

# After calling POST /send once, what is the value of results?

medium

A. ['Message sent']

B. [] (empty list)

C. ['Message sent', 'Message sent']

D. Error: background_tasks not defined

Solution

1. Step 1: Understand BackgroundTasks behavior

   BackgroundTasks run after the response is sent, but in testing or immediate inspection, the task may not have run yet.

2. Step 2: Check when results list updates

   The results list is updated inside the background task, which runs asynchronously after returning the response, so immediately after the call, results is still empty.

3. Final Answer:

   [] (empty list) -> Option B

4. Quick Check:

   Background task runs after response = empty results [OK]

Hint: BackgroundTasks run after response, so immediate results stay empty [OK]

Common Mistakes:

• Assuming background task runs before return
• Expecting results updated synchronously
• Confusing background_tasks parameter usage

4. Identify the error in this FastAPI code using BackgroundTasks:

from fastapi import FastAPI, BackgroundTasks
app = FastAPI()

@app.post('/notify')
async def notify():
    background_tasks.add_task(print, 'Notify event')
    return {'message': 'Notification sent'}

medium

A. BackgroundTasks must be imported from fastapi.background module

B. print function cannot be used in background tasks

C. Missing await keyword before add_task call

D. background_tasks parameter missing in function definition

Solution

1. Step 1: Check function parameters for BackgroundTasks

   The function uses background_tasks but does not declare it as a parameter, so it is undefined inside the function.

2. Step 2: Confirm correct usage of BackgroundTasks

   BackgroundTasks must be accepted as a parameter in the endpoint function to be used, e.g., 'async def notify(background_tasks: BackgroundTasks)'.

3. Final Answer:

   background_tasks parameter missing in function definition -> Option D

4. Quick Check:

   Missing parameter = background_tasks parameter missing in function definition [OK]

Hint: Always add BackgroundTasks as a function parameter to use it [OK]

Common Mistakes:

• Forgetting to add background_tasks parameter
• Thinking print can't be used in background tasks
• Adding await to add_task (it's not a coroutine)

5. You want to send an email notification after a user registers without delaying the response. Which FastAPI pattern best fits this event-driven need?

hard

A. Use BackgroundTasks to send the email after returning the response

B. Send the email synchronously inside the endpoint before returning

C. Use a global variable to store emails and send later manually

D. Call the email function inside a separate thread without FastAPI support

Solution

1. Step 1: Understand the goal of non-blocking email sending

   We want the user to get a quick response while the email sends in the background.

2. Step 2: Match FastAPI's event-driven feature

   BackgroundTasks lets you schedule tasks to run after the response, perfect for sending emails without delay.

3. Final Answer:

   Use BackgroundTasks to send the email after returning the response -> Option A

4. Quick Check:

   BackgroundTasks = non-blocking post-response tasks [OK]

Hint: Use BackgroundTasks for post-response actions like emails [OK]

Common Mistakes:

• Sending email synchronously causing delay
• Using global variables for task management
• Manually managing threads without FastAPI support