AIO-Databases

The package gives you async support for a range of databases (SQLite, PostgreSQL, MySQL).

Features

Has no dependencies (except databases drivers)
Supports asyncio and trio
Supports aiosqlite, aiomysql, aiopg, asyncpg, trio_mysql
Manage pools of connections
Manage transactions
Route read queries to replicas with write protection

Requirements

python >= 3.10

Installation

aio-databases should be installed using pip:

pip install aio-databases

You have to choose and install the required database drivers with:

# To support SQLite
pip install aio-databases[aiosqlite]  # asyncio

# To support MySQL
pip install aio-databases[aiomysql]   # asyncio
pip install aio-databases[trio_mysql] # trio

# To support PostgreSQL (choose one)
pip install aio-databases[aiopg]      # asyncio
pip install aio-databases[asyncpg]    # asyncio

# To support ODBC (alpha state)
pip install aio-databases[aioodbc]    # asyncio

Usage

Init a database

    from aio_databases import Database

    # Initialize a database
    db = Database('sqlite:///:memory:')  # with default driver

    # Flesh out the driver
    db = Database('asyncpg+pool://test:test@localhost:5432/tests', maxsize=10)

Supported schemas

aiomyql
aiomyql+pool
aiopg
aiopg+pool
asyncpg
asyncpg+pool
aioodbc
aioodbc+pool
aiosqlite
trio-mysql

Setup a pool of connections (optional)

Setup a pool of connections

    # Initialize a database's pool
    async def my_app_starts():
        await db.connect()

    # Close the pool
    async def my_app_ends():
        await db.disconnect()

    # As an alternative users are able to use the database
    # as an async context manager

    async with db:
        await my_main_coroutine()

Get a connection

    # Acquire and release (on exit) a connection
    async with db.connection():
        await my_code()

    # Acquire a connection only if it not exist
    async with db.connection(False):
        await my_code()

If a pool is setup it will be used

Run SQL queries

    await db.execute('select $1', '1')
    await db.executemany('select $1', '1', '2', '3')

    records = await db.fetchall('select (2 * $1) res', 2)
    assert records == [(4,)]

    record = await db.fetchone('select (2 * $1) res', 2)
    assert record == (4,)
    assert record['res'] == 4

    result = await db.fetchval('select 2 * $1', 2)
    assert result == 4

Iterate through rows one by one

    async for rec in db.iterate('select name from users'):
        print(rec)

Manage connections

By default the database opens and closes a connection for a query.

    # Connection will be acquired and released for the query
    await db.fetchone('select %s', 42)

    # Connection will be acquired and released again
    await db.fetchone('select %s', 77)

Manually open and close a connection

    # Acquire a new connection object
    async with db.connection():
        # Only one connection will be used
        await db.fetchone('select %s', 42)
        await db.fetchone('select %s', 77)
        # ...

    # Acquire a new connection or use an existing
    async with db.connection(False):
        # ...

If there any connection already db.method would be using the current one

    async with db.connection(): # connection would be acquired here
        await db.fetchone('select %s', 42)  # the connection is used
        await db.fetchone('select %s', 77)  # the connection is used

    # the connection released there

Reconnect

Pass reconnect=True to automatically drop a broken connection and acquire a fresh one:

    async with db.connection(reconnect=True):
        # If the connection dies, the query that hits the failure still raises,
        # but the broken connection is dropped and re-acquired eagerly.
        # The next query runs on the fresh connection.
        await db.fetchone('select %s', 42)

    # Or force a reconnect manually
    await conn.reconnect()

A failed query is never retried — the error propagates to the caller
The connection is re-acquired immediately after a connection error
Transactions are not restored: a transaction interrupted by a dead connection fails

Manage transactions

    # Start a tranction using the current connection
    async with db.transaction() as trans1:
        # do some work ...

        async with db.transaction() as trans2:
            # do some work ...
            await trans2.rollback()

        # unnessesary, the transaction will be commited on exit from the
        # current context

        await trans1.commit()

    # Create a new connection and start a transaction
    async with db.tranction(True) as trans:
        # do some work ...

Replicas

Configure replicas and route reads through them.

    db = Database(
        'asyncpg://primary/db',
        replicas=[
            'asyncpg://replica-1/db',
            'asyncpg://replica-2/db',
        ]
    )

Use db.replica() as an async context manager. Inside the block all queries run on a replica connection. execute and executemany raise ReadOnlyError, while fetch* queries work normally. Transactions are also blocked on replicas.

    async with db.replica():
        rows = await db.fetchall('select * from users')
        # raises ReadOnlyError:
        # await db.execute("insert into users ...")

Nested primary connections are allowed inside a replica block.

    async with db.replica():
        users = await db.fetchall('select * from users')

        async with db.connection():
            await db.execute('insert into users ...')

Bug tracker

If you have any suggestions, bug reports or annoyances please report them to the issue tracker at https://github.com/klen/aio-databases/issues

Contributing

Development of the project happens at: https://github.com/klen/aio-databases

License

Licensed under a MIT License

Release files for aio-databases 1.9.0

For a detailed explanation of source distributions (sdists) and built distributions (wheels), please see the package formats documentation.

Source distribution (sdist)

Source distribution for aio-databases 1.9.0
File	Size	Uploaded
aio_databases-1.9.0.tar.gz	12.2 kB	Sep 26, 2026	Details

Built distribution (wheel)

Table of built distributions (wheels) for aio-databases 1.9.0
File	Interpreter	ABI	Platform	Reset
aio_databases-1.9.0-py3-none-any.whl 18.3 kB Sep 26, 2026	Python 3	none	any	Details

Total release size: 30.6 kB

Release files / aio_databases-1.9.0.tar.gz

Download URL	aio_databases-1.9.0.tar.gz
Size	12.2 kB
Tags	Source
SHA-256 checksum How to use checksums	`93eb1ace516eb1c16b99156fbbbfbff7390ca882cddf3d94a2682b5fcc4d01ee`
BLAKE2b-256 checksum How to use checksums	`fc704e1c5454bda296d19c914896d78f348cfa88a361613e15cb1a0f804802c4`
Upload date	Sep 26, 2026
Uploaded using Trusted Publishing? What is trusted publishing?	No
Uploaded via	`uv/0.12.19 {"installer":{"name":"uv","version":"0.12.19","subcommand":["publish"]},"python":null,"implementation":{"name":null,"version":null},"distro":{"name":"Ubuntu","version":"24.04","id":"noble","libc":null},"system":{"name":null,"release":null},"cpu":null,"openssl_version":null,"setuptools_version":null,"rustc_version":null,"ci":true}`

Release files / aio_databases-1.9.0-py3-none-any.whl

Download URL	aio_databases-1.9.0-py3-none-any.whl
Size	18.3 kB
Tags	Python 3
SHA-256 checksum How to use checksums	`7fc1f61e47f66eea5829a58661767643794a373d4dce579be27b652a5d4d7753`
BLAKE2b-256 checksum How to use checksums	`5cefdb15fa0765bf2256d9f1db377b84670fbf77cc10f25590c6b0bc6b22030c`
Upload date	Sep 26, 2026
Uploaded using Trusted Publishing? What is trusted publishing?	No
Uploaded via	`uv/0.12.19 {"installer":{"name":"uv","version":"0.12.19","subcommand":["publish"]},"python":null,"implementation":{"name":null,"version":null},"distro":{"name":"Ubuntu","version":"24.04","id":"noble","libc":null},"system":{"name":null,"release":null},"cpu":null,"openssl_version":null,"setuptools_version":null,"rustc_version":null,"ci":true}`