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.

Practice

1. What is the main benefit of using async database queries in FastAPI?

easy

A. It removes the need for database connections.

B. It makes the database run faster internally.

C. The app can handle other tasks while waiting for the database response.

D. It automatically caches all database results.

Solution

1. Step 1: Understand async behavior in FastAPI

   Async lets the app continue running other code instead of waiting for slow operations like database queries.

2. Step 2: Identify the benefit of async database queries

   This means the app stays responsive and can handle more requests while waiting for the database.

3. Final Answer:

   The app can handle other tasks while waiting for the database response. -> Option C

4. Quick Check:

   Async = non-blocking tasks [OK]

Hint: Async means do other work while waiting [OK]

Common Mistakes:

• Thinking async speeds up the database itself
• Believing async caches results automatically
• Assuming async removes database connections

2. Which of the following is the correct way to define an async function for a database query in FastAPI?

easy

A. async def get_data(): return await session.execute(query)

B. async def get_data(): return session.execute(query)

C. def get_data(): return await session.execute(query)

D. def get_data(): return session.execute(query)

Solution

1. Step 1: Recognize async function syntax

   Async functions must be declared with async def to use await inside.

2. Step 2: Match correct usage of await

   To wait for the database query, await must be used with the async session call.

3. Final Answer:

   async def get_data(): return await session.execute(query) -> Option A

4. Quick Check:

   Async function + await = correct syntax [OK]

Hint: Use async def with await inside async functions [OK]

Common Mistakes:

• Missing async keyword but using await
• Using await without async function
• Calling async functions without await

3. Given this async function in FastAPI using SQLAlchemy async session:

async def fetch_users(session):
    result = await session.execute("SELECT name FROM users")
    users = result.scalars().all()
    return users

What will fetch_users return?

medium

A. A coroutine object that needs awaiting

B. A list of user names from the database

C. A SQLAlchemy Result object

D. An error because scalars() is not async

Solution

1. Step 1: Understand await on session.execute

   The await waits for the query to finish and returns a Result object.

2. Step 2: Use scalars().all() on Result

   Calling scalars().all() extracts all values from the first column as a list.

3. Final Answer:

   A list of user names from the database -> Option B

4. Quick Check:

   await + scalars().all() = list output [OK]

Hint: await session.execute returns Result; scalars().all() gives list [OK]

Common Mistakes:

• Thinking fetch_users returns a coroutine, not awaited
• Expecting a Result object instead of list
• Assuming scalars() is async and causes error

4. Identify the error in this FastAPI async database query code:

async def get_items(session):
    result = session.execute("SELECT * FROM items")
    return await result.fetchall()

medium

A. Function should not be async

B. Cannot await fetchall() method

C. session.execute should not be async

D. Missing await before session.execute call

Solution

1. Step 1: Check usage of async session.execute

   session.execute is an async method and must be awaited to get the Result.

2. Step 2: Understand fetchall() usage

   fetchall() is called on the awaited Result; awaiting fetchall() is incorrect because it returns a list synchronously.

3. Final Answer:

   Missing await before session.execute call -> Option D

4. Quick Check:

   await session.execute needed before fetchall() [OK]

Hint: Always await async session.execute calls first [OK]

Common Mistakes:

• Not awaiting session.execute causing coroutine error
• Awaiting fetchall() which is not async
• Confusing async with sync methods

5. You want to fetch all active users asynchronously and return their emails in FastAPI. Which code snippet correctly uses async database queries and filters results?

hard

A. async def get_active_emails(session):
result = await session.execute("SELECT email FROM users WHERE active=1")
return result.scalars().all()

B. def get_active_emails(session):
result = session.execute("SELECT email FROM users WHERE active=1")
return result.scalars().all()

C. async def get_active_emails(session):
result = session.execute("SELECT email FROM users WHERE active=1")
return await result.scalars().all()

D. async def get_active_emails(session):
result = await session.execute("SELECT email FROM users")
return [email for email in result.scalars().all() if email.active == 1]

Solution

1. Step 1: Use async def and await session.execute

   The function must be async and await the async session.execute call with the correct SQL filter.

2. Step 2: Extract emails with scalars().all()

   Calling scalars().all() on the awaited result returns a list of emails matching the filter.

3. Final Answer:

   async def get_active_emails(session): result = await session.execute("SELECT email FROM users WHERE active=1") return result.scalars().all() -> Option A

4. Quick Check:

   Async + await + filtered query = correct [OK]

Hint: Await session.execute with filter, then scalars().all() [OK]

Common Mistakes:

• Using sync def with async calls
• Not awaiting session.execute
• Filtering results in Python instead of SQL