Async database queries in FastAPI - Practice Problems & Coding Challenges

Challenge - 5 Problems

❓ component_behavior

intermediate

What is the output of this async FastAPI endpoint?

Consider this FastAPI endpoint using async SQLAlchemy to fetch a user by ID. What will the endpoint return if the user with ID 5 exists in the database?

FastAPI

from fastapi import FastAPI, HTTPException
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy.future import select
from models import User

app = FastAPI()

@app.get('/users/{user_id}')
async def get_user(user_id: int, session: AsyncSession):
    result = await session.execute(select(User).where(User.id == user_id))
    user = result.scalars().first()
    if user is None:
        raise HTTPException(status_code=404, detail='User not found')
    return {'id': user.id, 'name': user.name}

A HTTP 500 Internal Server Error

B {"detail": "User not found"}

C {"id": 5, "name": "Alice"}

D None

💡 Hint

Think about what happens when the user exists and how the code returns the user data.

The code queries asynchronously for a user with the given ID. If found, it returns a JSON with the user's id and name. Since user 5 exists, it returns their data.

📝 Syntax

intermediate

Which option correctly awaits an async database query in FastAPI?

You want to fetch all items asynchronously from the database using SQLAlchemy in a FastAPI endpoint. Which code snippet correctly awaits the query?

FastAPI

from sqlalchemy.future import select
from models import Item

async def get_items(session):
    # Fill in the correct await syntax here
    items = ???
    return items

A items = await session.execute(select(Item))

B items = session.execute(await select(Item))

C items = session.execute(select(Item))

D items = await select(Item)

💡 Hint

Remember that the session.execute method is async and must be awaited.

The session.execute method is async and requires awaiting. The select() function is synchronous and returns a query object, so it should not be awaited.

🔧 Debug

advanced

Why does this async FastAPI endpoint raise a RuntimeError?

Examine this FastAPI endpoint code. It raises 'RuntimeError: This event loop is already running' when called. What causes this error?

FastAPI

from fastapi import FastAPI
import asyncio

app = FastAPI()

@app.get('/data')
def get_data():
    loop = asyncio.get_event_loop()
    result = loop.run_until_complete(fetch_data())
    return result

async def fetch_data():
    return {'data': 123}

A Calling run_until_complete inside an already running event loop causes RuntimeError.

B fetch_data is not awaited properly because it is not async.

C FastAPI does not support async endpoints, causing the error.

D The event loop is missing and must be created manually.

💡 Hint

FastAPI runs on an event loop already. Using run_until_complete inside it causes conflict.

FastAPI runs on an async event loop. Calling run_until_complete inside an already running loop causes a RuntimeError.

❓ state_output

advanced

What is the value of 'count' after running this async database query loop?

Given this async function that counts users with age > 30, what is the final value of 'count'?

FastAPI

async def count_users(session):
    count = 0
    result = await session.stream(select(User).where(User.age > 30))
    async for user in result.scalars():
        count += 1
    return count

A 0, because scalars() cannot be iterated asynchronously

B The total number of users regardless of age

C Raises a TypeError due to incorrect async iteration

D The number of users in the database with age greater than 30

💡 Hint

The code uses async iteration over the query result scalars filtered by age.

The async for loop iterates over users with age > 30, incrementing count for each. So count equals the number of such users.

🧠 Conceptual

expert

Which statement best explains why async database queries improve FastAPI performance?

Why do async database queries help FastAPI handle more requests efficiently compared to synchronous queries?

A Async queries run faster because they use multiple CPU cores automatically.

B Async queries free the event loop during I/O waits, allowing other requests to run concurrently.

C Async queries cache all data in memory, reducing database load.

D Async queries block the event loop less by running queries in separate threads.

💡 Hint

Think about what happens when the app waits for the database response.

Async queries let FastAPI pause waiting for database I/O, so it can handle other requests without blocking.