Async database with databases library in FastAPI

Introduction

Using async database calls lets your app handle many requests smoothly without waiting for the database. The databases library helps you do this easily with async support.

When building a web app that needs to handle many users at the same time.

When you want your app to stay fast even if database queries take time.

When using FastAPI or other async frameworks that support async code.

When you want simple code to connect and query databases asynchronously.

Syntax

FastAPI

import databases

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

async def connect_db():
    await database.connect()

async def disconnect_db():
    await database.disconnect()

async def fetch_data():
    query = "SELECT * FROM tablename"
    results = await database.fetch_all(query=query)
    return results

Use await to wait for database operations without blocking.

Call connect() before queries and disconnect() when done.

Examples

Example connecting to a PostgreSQL database asynchronously.

FastAPI

import databases

DATABASE_URL = "postgresql+asyncpg://user:pass@localhost/dbname"
database = databases.Database(DATABASE_URL)

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

Example inserting data asynchronously with parameters.

FastAPI

async def add_user(name: str):
    query = "INSERT INTO users(name) VALUES (:name)"
    await database.execute(query=query, values={"name": name})

Sample Program

This example creates a simple notes table, inserts one note, then fetches and prints all notes asynchronously using the databases library with SQLite.

FastAPI

import databases
import sqlalchemy
import asyncio

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

metadata = sqlalchemy.MetaData()

notes = sqlalchemy.Table(
    "notes",
    metadata,
    sqlalchemy.Column("id", sqlalchemy.Integer, primary_key=True),
    sqlalchemy.Column("text", sqlalchemy.String),
)

engine = sqlalchemy.create_engine(
    DATABASE_URL.replace("+aiosqlite", ""), connect_args={"check_same_thread": False}
)
metadata.create_all(engine)

async def main():
    await database.connect()

    # Insert a note
    query = notes.insert().values(text="Remember to learn async DB calls!")
    await database.execute(query)

    # Fetch notes
    query = notes.select()
    rows = await database.fetch_all(query)

    for row in rows:
        print(f"Note {row['id']}: {row['text']}")

    await database.disconnect()

asyncio.run(main())

Important Notes

Always call await database.connect() before running queries.

Use parameterized queries to avoid SQL injection risks.

SQLite needs special connection args for async use with aiosqlite.

Summary

The databases library helps run database queries asynchronously in FastAPI.

Async calls keep your app responsive by not blocking while waiting for the database.

Remember to connect before queries and disconnect when done.