Async database queries in FastAPI - Mini Project: Build & Apply

Async database queries with FastAPI

📖 Scenario: You are building a simple FastAPI app that fetches user data from a database asynchronously. This helps your app stay fast and responsive even when waiting for the database.

🎯 Goal: Create a FastAPI app that uses async functions to query a mock database and return user info.

📋 What You'll Learn

Create a list of user dictionaries as mock database

Add a variable to hold the user ID to search for

Write an async function to find the user by ID

Create a FastAPI route that calls the async function and returns the user

💡 Why This Matters

🌍 Real World

Async database queries keep web apps responsive by not blocking the server while waiting for data. This is important for apps with many users or slow databases.

💼 Career

Understanding async queries and FastAPI routes is essential for backend developers building modern, scalable web APIs.

1

Create mock user data

Create a list called users with these exact dictionaries: {'id': 1, 'name': 'Alice'}, {'id': 2, 'name': 'Bob'}, and {'id': 3, 'name': 'Charlie'}.

FastAPI

# Create the list called users with 3 user dictionaries
# Your code here

Hint

Use a list with dictionaries inside. Each dictionary has keys 'id' and 'name'.

2

Set user ID to search

Create a variable called search_id and set it to 2.

FastAPI

users = [
    {'id': 1, 'name': 'Alice'},
    {'id': 2, 'name': 'Bob'},
    {'id': 3, 'name': 'Charlie'}
]
# Create search_id variable and set to 2
# Your code here

Hint

Just assign the number 2 to the variable named search_id.

3

Write async function to find user

Write an async function called get_user_by_id that takes user_id as a parameter. Inside, use a for loop with user to check if user['id'] == user_id. If yes, return user. If no user found, return None.

FastAPI

users = [
    {'id': 1, 'name': 'Alice'},
    {'id': 2, 'name': 'Bob'},
    {'id': 3, 'name': 'Charlie'}
]

search_id = 2

# Write async function get_user_by_id(user_id) to find user
# Your code here

Hint

Use async def to define the function. Loop through users and compare IDs.

4

Create FastAPI route to return user

Import FastAPI from fastapi. Create an app instance called app. Add a route with @app.get('/users/{user_id}') that defines an async function read_user taking user_id: int. Inside, call get_user_by_id(user_id) with await and return the result.

FastAPI

users = [
    {'id': 1, 'name': 'Alice'},
    {'id': 2, 'name': 'Bob'},
    {'id': 3, 'name': 'Charlie'}
]

search_id = 2

async def get_user_by_id(user_id):
    for user in users:
        if user['id'] == user_id:
            return user
    return None

# Import FastAPI, create app, and add async route /users/{user_id}
# Your code here

Hint

Remember to import FastAPI, create app, and use async def with await inside the route.

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