Event-driven architecture in FastAPI - Interactive Code Practice

Practice - 5 Tasks

1fill in blank

easy

Complete the code to define an event handler function in FastAPI.

FastAPI

from fastapi import FastAPI

app = FastAPI()

@app.[1]("/items/")
async def create_item():
    return {"message": "Item created"}

Drag options to blanks, or click blank then click option'

Aget

Bpost

Cput

Ddelete

Attempts:

3 left

💡 Hint

Common Mistakes

Using @app.get instead of @app.post for creating resources.

Confusing HTTP methods for event handling.

The @app.post decorator defines a POST request handler, which is commonly used to create resources in event-driven APIs.

2fill in blank

medium

Complete the code to publish an event message asynchronously.

FastAPI

import asyncio

async def publish_event(event_data):
    await [1](event_data)

async def send_message(data):
    print(f"Sending: {data}")

Drag options to blanks, or click blank then click option'

Asend_message

Bdispatch

Cemit

Dpublish

Attempts:

3 left

💡 Hint

Common Mistakes

Using undefined functions like publish or emit without definition.

Not awaiting the asynchronous function.

The function send_message is awaited to simulate sending an event asynchronously.

3fill in blank

hard

Fix the error in the event listener registration code.

FastAPI

from fastapi import FastAPI

app = FastAPI()

@[1]("/event")
async def event_listener():
    return {"status": "Event received"}

Drag options to blanks, or click blank then click option'

Aapp.get

Bapp.listen

Capp.event

Dapp.on

Attempts:

3 left

💡 Hint

Common Mistakes

Using non-existent decorators like app.listen or app.on.

Confusing event-driven frameworks with FastAPI syntax.

FastAPI uses @app.get or other HTTP method decorators to define endpoints. app.listen or app.on are not valid decorators.

4fill in blank

hard

Fill both blanks to create a background task that handles events.

FastAPI

from fastapi import FastAPI, BackgroundTasks

app = FastAPI()

def process_event(data):
    print(f"Processing {data}")

@app.post("/event")
async def handle_event(data: dict, [1]: BackgroundTasks):
    [2].add_task(process_event, data)
    return {"message": "Event received"}

Drag options to blanks, or click blank then click option'

Abackground_tasks

Btasks

Cbackground

Dbg_tasks

Attempts:

3 left

💡 Hint

Common Mistakes

Using different variable names for the parameter and method call.

Not using BackgroundTasks for background processing.

The parameter name background_tasks is used to receive the BackgroundTasks instance, and then background_tasks.add_task schedules the event processing.

5fill in blank

hard

Fill all three blanks to implement a simple event bus with subscription and publishing.

FastAPI

class EventBus:
    def __init__(self):
        self.subscribers = {}

    def subscribe(self, event_type, handler):
        if event_type not in self.subscribers:
            self.subscribers[[1]] = []
        self.subscribers[event_type].append(handler)

    async def publish(self, event_type, data):
        for handler in self.subscribers.get(event_type, [2]):
            await handler([3])

Drag options to blanks, or click blank then click option'

Aevent_type

B[]

Cdata

Dhandler

Attempts:

3 left

💡 Hint

Common Mistakes

Using wrong keys for the subscribers dictionary.

Not providing a default empty list in get method.

Passing wrong argument to handlers.

We use event_type as the key to store handlers, default to empty list [] if no subscribers, and pass data to each handler when publishing.