databases

This example demonstrates how to initialize a connection, create a table, insert multiple rows, and fetch results using aiosqlite.

# Create a database instance, and connect to it.
from databases import Database
database = Database('sqlite+aiosqlite:///example.db')
await database.connect()

# Create a table.
query = """CREATE TABLE HighScores (id INTEGER PRIMARY KEY, name VARCHAR(100), score INTEGER)"""
await database.execute(query=query)

# Insert some data.
query = "INSERT INTO HighScores(name, score) VALUES (:name, :score)"
values = [
    {"name": "Daisy", "score": 92},
    {"name": "Neil", "score": 87},
    {"name": "Carol", "score": 43},
]
await database.execute_many(query=query, values=values)

# Run a database query.
query = "SELECT * FROM HighScores"
rows = await database.fetch_all(query=query)
print('High Scores:', rows)

You can manage the database connection pool using an async context manager, which automatically handles connecting and disconnecting. Alternatively, you can use explicit .connect() and .disconnect() methods. If integrating with a web framework like Starlette, it is recommended to hook these methods into the framework's startup and shutdown events.

# Using an async context manager
async with Database(DATABASE_URL) as database:
    ...

# Using explicit connect/disconnect
database = Database(DATABASE_URL)
await database.connect()
...
await database.disconnect()

Install the core databases package using pip:

$ pip install databases

To use SQLAlchemy core queries, you must declare your tables in your code using sqlalchemy.MetaData and sqlalchemy.Table. This allows you to keep your database schema in sync with your application code and enables the use of migration tools. You can use standard SQLAlchemy column types like sqlalchemy.Integer, sqlalchemy.String, sqlalchemy.Boolean, or sqlalchemy.JSON.

import sqlalchemy

metadata = sqlalchemy.MetaData()

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

Nested transactions are supported and are implemented using database savepoints. If an inner transaction fails (e.g., by raising an exception), you can suppress that exception to prevent it from affecting the outer transaction, allowing the outer transaction to continue.

import contextlib

async with databases.Database(database_url) as db:
    async with db.transaction() as outer:
        # Do something in the outer transaction
        ...

        # Suppress to prevent influence on the outer transaction
        with contextlib.suppress(ValueError):
            async with db.transaction():
                # Do something in the inner transaction
                ...
                raise ValueError('Abort the inner transaction')

    # Observe the results of the outer transaction, without effects from the inner transaction.
    await db.fetch_all('SELECT * FROM ...')

Install the specific async driver required for your database type using the following extras:

• PostgreSQL: databases[asyncpg] or databases[aiopg]
• MySQL: databases[aiomysql] or databases[asyncmy]
• SQLite: databases[aiosqlite]

Note: If you are using synchronous SQLAlchemy functions (like engine.create_all()) or Alembic migrations, you must also install a synchronous driver: psycopg2 for PostgreSQL or pymysql for MySQL.

$ pip install databases[asyncpg]
$ pip install databases[aiopg]
$ pip install databases[aiomysql]
$ pip install databases[asyncmy]
$ pip install databases[aiosqlite]

To ensure strict test isolation, you can configure the Database instance to run all connections within a transaction that automatically rolls back when the database is disconnected. This prevents test data from persisting between test cases.

When integrating with a web framework, use a conditional check to use a dedicated test database URL with force_rollback=True during testing.

# For strict isolation in all cases
database = Database(DATABASE_URL, force_rollback=True)

# Typical pattern for web frameworks
if TESTING:
    database = Database(TEST_DATABASE_URL, force_rollback=True)
else:
    database = Database(DATABASE_URL)

PostgreSQL and MySQL backends allow configuring SSL and connection pool sizes via the connection URL query parameters or as keyword arguments in the Database constructor. Common options include ssl and pool size settings like min_size and max_size.

# Using URL query parameters
database = Database('postgresql+asyncpg://localhost/example?ssl=true')
database = Database('mysql+aiomysql://localhost/example?min_size=5&max_size=20')

# Using keyword arguments
database = Database('postgresql+asyncpg://localhost/example', ssl=True, min_size=5, max_size=20)

The Database class provides the following primary async methods for interacting with your database:

• await database.connect(): Establishes the connection to the database.
• await database.execute(query=...): Executes a single query (e.g., CREATE TABLE, INSERT).
• await database.execute_many(query=..., values=...): Executes a query against multiple sets of values (bulk insert).
• await database.fetch_all(query=...): Executes a query and returns all resulting rows.