Async database with databases library in FastAPI - Practice Problems & Coding Challenges

Challenge - 5 Problems

❓ component_behavior

intermediate

What is the output of this FastAPI async database query?

Consider this FastAPI endpoint using the databases library to fetch all users asynchronously. What will the response contain?

FastAPI

from fastapi import FastAPI
import databases

DATABASE_URL = "sqlite+aiosqlite:///./test.db"
database = databases.Database(DATABASE_URL)
app = FastAPI()

@app.on_event("startup")
async def startup():
    await database.connect()

@app.on_event("shutdown")
async def shutdown():
    await database.disconnect()

@app.get("/users")
async def read_users():
    query = "SELECT * FROM users"
    results = await database.fetch_all(query=query)
    return results

A An empty list regardless of database content

B A single dictionary representing the first user row only

C A runtime error because fetch_all is not awaited

D A list of dictionaries, each representing a user row from the users table

💡 Hint

Remember that fetch_all returns all matching rows asynchronously.

The fetch_all method returns a list of rows matching the query. Since the query selects all users, the response is a list of dictionaries, each dictionary representing a user.

📝 Syntax

intermediate

Which option correctly initializes the databases.Database instance for PostgreSQL?

You want to connect asynchronously to a PostgreSQL database using the databases library. Which DATABASE_URL string is correct?

A "postgresql://user:password@localhost/dbname"

B "postgresql+psycopg2://user:password@localhost/dbname"

C "postgresql+asyncpg://user:password@localhost/dbname"

D "postgresql+aiopg://user:password@localhost/dbname"

💡 Hint

The databases library requires an async driver prefix for async support.

For async PostgreSQL connections with the databases library, the URL must use an async driver like asyncpg. So the correct prefix is postgresql+asyncpg://.

🔧 Debug

advanced

Why does this FastAPI app raise an error on startup with databases library?

Given this code snippet, why does the app fail to start with an error related to database connection?

FastAPI

from fastapi import FastAPI
import databases

DATABASE_URL = "sqlite+aiosqlite:///./test.db"
database = databases.Database(DATABASE_URL)
app = FastAPI()

@app.on_event("startup")
async def startup():
    await database.connect()

@app.on_event("shutdown")
async def shutdown():
    await database.disconnect()

A The startup event handler is not async, so await is missing causing a runtime error

B The database URL is invalid for SQLite

C The shutdown event handler should not be async

D The database instance is not created correctly

💡 Hint

Check if the database connect method is awaited properly.

The database.connect() method is async and must be awaited. The startup event handler is defined as a normal function, so it cannot await. This causes a runtime error.

❓ state_output

advanced

What is the value of 'user_count' after this async query?

Assuming the users table has 5 rows, what will be the value of user_count after running this code?

FastAPI

query = "SELECT COUNT(*) FROM users"
user_count = await database.fetch_val(query)

A 5

B [5]

C {'count': 5}

D None

💡 Hint

fetch_val returns the first column of the first row.

The fetch_val method returns the value of the first column in the first row of the query result. Since the query counts users, it returns the integer 5.

🧠 Conceptual

expert

Which statement about transaction management with databases library is correct?

Select the correct statement about handling transactions asynchronously using the databases library in FastAPI.

A Transactions are automatically committed after each query without explicit control

B You can use 'async with database.transaction():' to run multiple queries atomically

C The databases library does not support transactions in async mode

D You must manually call database.commit() after each query to save changes

💡 Hint

Think about Python's async context managers for transactions.

The databases library supports async transactions using an async context manager. Using async with database.transaction(): ensures all queries inside run atomically and commit or rollback together.

Practice

1. What is the main benefit of using the databases library with FastAPI for database operations?

easy

A. It automatically creates database tables without any code.

B. It allows running database queries asynchronously, keeping the app responsive.

C. It replaces FastAPI's routing system with database queries.

D. It makes the database run faster by caching all queries.

Solution

1. Step 1: Understand asynchronous behavior in FastAPI

   FastAPI supports async functions to avoid blocking operations, improving responsiveness.

2. Step 2: Role of the databases library

   The databases library allows async database queries, so the app doesn't wait for the database to respond before continuing.

3. Final Answer:

   It allows running database queries asynchronously, keeping the app responsive. -> Option B

4. Quick Check:

   Async queries keep app responsive = A [OK]

Hint: Async means non-blocking queries for smooth app flow [OK]

Common Mistakes:

• Thinking it auto-creates tables
• Confusing async with caching
• Believing it changes routing

2. Which of the following is the correct way to connect to the database using the databases library in FastAPI?

easy

A. await database.connection()

B. database.connect()

C. database.await_connect()

D. await database.connect()

Solution

1. Step 1: Recall async connection method

   The databases library requires awaiting the connect method because it is asynchronous.

2. Step 2: Identify correct syntax

   The correct syntax is await database.connect(). Calling without await or wrong method names causes errors.

3. Final Answer:

   await database.connect() -> Option D

4. Quick Check:

   Async connect uses await = A [OK]

Hint: Always await async connect calls in databases library [OK]

Common Mistakes:

• Forgetting to use await
• Using wrong method names
• Calling connect synchronously

3. Given this FastAPI code snippet using the databases library, what will be printed when the endpoint is called?

import databases
from fastapi import FastAPI

database = databases.Database('sqlite:///test.db')
app = FastAPI()

@app.on_event('startup')
async def startup():
    await database.connect()

@app.on_event('shutdown')
async def shutdown():
    await database.disconnect()

@app.get('/')
async def read_data():
    query = 'SELECT 1 as number'
    result = await database.fetch_one(query)
    print(result['number'])
    return {'number': result['number']}

medium

A. Error: invalid query

B. None

C. 1

D. Error: database not connected

Solution

1. Step 1: Check database connection lifecycle

   The database connects on startup and disconnects on shutdown, so it is connected when the endpoint runs.

2. Step 2: Analyze query and fetch_one result

   The query selects the number 1 as 'number'. The fetch_one returns a dict-like object with key 'number' and value 1.

3. Final Answer:

   1 -> Option C

4. Quick Check:

   Query returns 1 = D [OK]

Hint: Check connection events and query result keys [OK]

Common Mistakes:

• Assuming no connection established
• Expecting None instead of 1
• Misreading query syntax

4. Identify the error in this FastAPI code using the databases library:

import databases
from fastapi import FastAPI

database = databases.Database('sqlite:///test.db')
app = FastAPI()

@app.on_event('startup')
async def startup():
    await database.connect()

@app.on_event('shutdown')
async def shutdown():
    await database.disconnect()

@app.get('/')
async def read_data():
    query = 'SELECT 1 as number'
    result = database.fetch_one(query)
    return {'number': result['number']}

medium

A. Missing await before database.fetch_one causing a runtime error

B. Database URL is incorrect

C. Missing database.connect() call

D. Using synchronous FastAPI endpoint instead of async

Solution

1. Step 1: Check usage of async database method

   The fetch_one method is async and must be awaited to get the result properly.

2. Step 2: Identify missing await

   Code calls database.fetch_one(query) without await, which causes a runtime error because the coroutine is not awaited.

3. Final Answer:

   Missing await before database.fetch_one causing a runtime error -> Option A

4. Quick Check:

   Async calls need await = C [OK]

Hint: Always await async database calls to avoid errors [OK]

Common Mistakes:

• Forgetting await on async calls
• Assuming fetch_one is synchronous
• Ignoring connection lifecycle

5. You want to fetch all users from a database asynchronously using the databases library in FastAPI. Which code snippet correctly fetches all rows and returns them as a list of dictionaries?

database = databases.Database('sqlite:///test.db')

async def get_users():
    query = 'SELECT * FROM users'
    # Which line correctly fetches all rows?
    ???
    return users

hard

A. users = await database.fetch_all(query)

B. users = database.fetch_all(query)

C. users = await database.fetch_one(query)

D. users = database.fetch_one(query)

Solution

1. Step 1: Understand fetch_all vs fetch_one

   fetch_all returns all rows as a list; fetch_one returns a single row.

2. Step 2: Use await with async fetch_all

   Since fetch_all is async, it must be awaited to get the result.

3. Final Answer:

   users = await database.fetch_all(query) -> Option A

4. Quick Check:

   Fetch all rows async with await = B [OK]

Hint: Use await with fetch_all to get all rows asynchronously [OK]

Common Mistakes:

• Using fetch_one to get all rows
• Forgetting await on async calls
• Calling fetch_all synchronously