Async database with databases library in FastAPI - Mini Project: Build & Apply

Async database with databases library

📖 Scenario: You are building a simple FastAPI app that connects to a SQLite database asynchronously using the databases library. This app will store and retrieve user names.

🎯 Goal: Create a FastAPI app that connects asynchronously to a SQLite database using the databases library. You will define the database URL, create a database instance, write an async function to fetch all users, and finally add the startup and shutdown events to connect and disconnect the database.

📋 What You'll Learn

Use the databases library to handle async database connections

Use SQLite as the database with the URL sqlite:///./test.db

Create a database variable for the connection

Write an async function fetch_users that fetches all users from a users table

Add FastAPI startup and shutdown event handlers to connect and disconnect the database

💡 Why This Matters

🌍 Real World

Many web applications need to access databases asynchronously to handle many users efficiently without blocking the server.

💼 Career

Knowing how to use async database libraries like databases with FastAPI is valuable for backend developer roles focused on modern Python web frameworks.

1

Set up the database URL and import libraries

Import databases and FastAPI. Create a variable called DATABASE_URL and set it to "sqlite:///./test.db".

FastAPI

# Your code here

Need a hint?

Use import databases and from fastapi import FastAPI. Then assign the exact string "sqlite:///./test.db" to DATABASE_URL.

2

Create the database instance

Create a variable called database and assign it to databases.Database(DATABASE_URL).

FastAPI

from fastapi import FastAPI
import databases

DATABASE_URL = "sqlite:///./test.db"

# Create the database instance
# Your code here

Need a hint?

Use databases.Database with the DATABASE_URL variable to create the database instance.

3

Write async function to fetch all users

Define an async function called fetch_users that returns the result of await database.fetch_all(query). Use the query "SELECT * FROM users" inside the function.

FastAPI

from fastapi import FastAPI
import databases

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

# Define async function to fetch users
# Your code here

Need a hint?

Define an async function named fetch_users. Inside, set query to the exact string "SELECT * FROM users" and return await database.fetch_all(query).

4

Add FastAPI app and connect/disconnect events

Create a FastAPI app instance called app. Add @app.on_event("startup") decorated async function startup that calls await database.connect(). Add @app.on_event("shutdown") decorated async function shutdown that calls await database.disconnect().

FastAPI

from fastapi import FastAPI
import databases

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

async def fetch_users():
    query = "SELECT * FROM users"
    return await database.fetch_all(query)

# Create FastAPI app and add startup/shutdown events
# Your code here

Need a hint?

Create app = FastAPI(). Use @app.on_event("startup") to decorate an async startup function that calls await database.connect(). Similarly, use @app.on_event("shutdown") to decorate an async shutdown function that calls await database.disconnect().