Async database with databases library in FastAPI - Interactive Code Practice

Practice - 5 Tasks

1fill in blank

easy

Complete the code to import the async database library.

FastAPI

from databases import [1]

Drag options to blanks, or click blank then click option'

Adatabase

BDatabase

CDatabases

Ddb

Attempts:

3 left

💡 Hint

Common Mistakes

Using lowercase 'database' instead of 'Database'.

Trying to import a non-existent module name.

The Database class is imported from the databases library to create an async database connection.

2fill in blank

medium

Complete the code to create a database connection with the URL.

FastAPI

database = Database([1])

Drag options to blanks, or click blank then click option'

A'postgresql://user:pass@localhost/dbname'

B'mysql://user:pass@localhost/dbname'

C'mongodb://localhost:27017'

D'sqlite:///test.db'

Attempts:

3 left

💡 Hint

Common Mistakes

Using SQLite URL when expecting PostgreSQL.

Using MongoDB URL which is not supported by this library.

The connection string for PostgreSQL uses the format postgresql://user:pass@host/dbname. This is the correct URL for the Database constructor.

3fill in blank

hard

Fix the error in the async function to connect to the database.

FastAPI

async def connect_db():
    await database.[1]()

Drag options to blanks, or click blank then click option'

Aopen

Bstart

Cconnect

Dconnect_db

Attempts:

3 left

💡 Hint

Common Mistakes

Using a non-existent method like 'start' or 'open'.

Calling a method that does not exist on the database object.

The Database class uses the connect() method to open the connection asynchronously.

4fill in blank

hard

Fill both blanks to create a query and fetch all rows asynchronously.

FastAPI

query = "SELECT * FROM users WHERE id = [1]"
rows = await database.[2](query)

Drag options to blanks, or click blank then click option'

A:user_id

Bfetch_all

Cexecute

D?user_id

Attempts:

3 left

💡 Hint

Common Mistakes

Using '?' instead of ':user_id' for named parameters.

Using 'execute' which runs the query but does not fetch results.

The query uses a named parameter :user_id for safe substitution. The fetch_all method retrieves all matching rows asynchronously.

5fill in blank

hard

Fill all three blanks to insert a new user asynchronously and get the inserted ID.

FastAPI

query = "INSERT INTO users(name, email) VALUES ([1], [2]) RETURNING id"
values = {"name": user_name, "email": user_email}
user_id = await database.[3](query=query, values=values)

Drag options to blanks, or click blank then click option'

A:name

B:email

Cexecute

Dfetch_one

Attempts:

3 left

💡 Hint

Common Mistakes

Using 'execute' which runs the insert but does not return the ID for PostgreSQL.

Using positional parameters instead of named parameters.

The query uses named parameters :name and :email for safe insertion. Include RETURNING id (for PostgreSQL) and use the fetch_one method to retrieve the inserted record's ID asynchronously. The result is a dict like {'id': 1}.

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