Async database with databases library in FastAPI - Step-by-Step Execution

Concept Flow - Async database with databases library

Start FastAPI app

↓

Create Database instance

↓

Connect to DB asynchronously

↓

Handle API request

↓

Run async DB query

↓

Return query result

↓

Disconnect from DB on shutdown

This flow shows how FastAPI app uses the databases library to connect asynchronously to a database, run queries on requests, and disconnect cleanly.

Execution Sample

FastAPI

from fastapi import FastAPI
import databases

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

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

This code creates a FastAPI app and connects asynchronously to a SQLite database on startup.

Execution Table

Step	Action	Async Operation	State Before	State After	Output/Result
1	Create FastAPI app	No	No app	app instance created	App ready
2	Create Database instance	No	No database object	database object created	DB connection string set
3	Startup event triggered	Yes	database disconnected	database connected	DB connection established
4	API request received	No	database connected	database connected	Request processing starts
5	Run async query	Yes	database connected	database connected	Query result fetched
6	Return response	No	database connected	database connected	Response sent to client
7	Shutdown event triggered	Yes	database connected	database disconnected	DB connection closed
8	App stopped	No	database disconnected	No app running	App shutdown complete

💡 App stops after shutdown event closes the database connection.

Variable Tracker

Variable	Start	After Step 2	After Step 3	After Step 5	After Step 7	Final
app	None	FastAPI instance	FastAPI instance	FastAPI instance	FastAPI instance	None (app stopped)
database	None	Database instance (disconnected)	Database connected	Database connected	Database disconnected	None (closed)
query_result	None	None	None	Data from DB	Data from DB	None

Key Moments - 3 Insights

Why do we use 'await' before database.connect()?

Because database.connect() is an async operation that takes time, 'await' pauses execution until connection completes, as shown in step 3 of the execution_table.

What happens if we try to query the database before connecting?

The query will fail because the database is not connected yet. Step 3 ensures connection before queries run in step 5.

Why disconnect the database on shutdown?

To free resources and avoid connection leaks. Step 7 shows the async disconnect operation during app shutdown.

Visual Quiz - 3 Questions

Look at the execution_table, what is the state of 'database' after step 3?

A Disconnected

B Connected

C Not created

D Closed

💡 Hint

Check the 'State After' column for step 3 in the execution_table.

At which step does the app send the query result back to the client?

A Step 4

B Step 5

C Step 6

D Step 7

💡 Hint

Look for 'Return response' and 'Response sent to client' in the execution_table.

If we omit 'await' before database.connect(), what will happen?

A The app will raise an error or not wait for connection

B The database disconnects immediately

C The connection will still work synchronously

D The query runs before connection

💡 Hint

Recall that step 3 uses 'await' to pause until connection completes.

Concept Snapshot

Async database with databases library in FastAPI:
- Create Database instance with connection string
- Use async def startup() with await database.connect()
- Run async queries with await database.fetch_all() etc.
- Disconnect on shutdown with await database.disconnect()
- Always use 'await' for async DB calls to avoid errors

Full Transcript

This visual execution shows how to use the databases library asynchronously with FastAPI. First, the FastAPI app and a Database instance are created. On startup, the app connects asynchronously to the database using await database.connect(). When an API request comes in, the app runs async queries with await, fetching data without blocking. After sending the response, on shutdown the app disconnects asynchronously to clean up. Variables like 'database' change state from disconnected to connected and back. Key points include always awaiting async calls and connecting before querying. The execution table traces each step from app creation to shutdown, showing state changes and outputs clearly.

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