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
aiomyqlaiomyql+poolaiopgaiopg+poolasyncpgasyncpg+poolaioodbcaioodbc+poolaiosqlitetrio-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
1 maintainer
Metadata
1 maintainer
Release files for aio-databases 2.0.0
For a detailed explanation of source distributions (sdists) and built distributions (wheels), please see the package formats documentation.
Source distribution (sdist)
| File | Size | Uploaded | Provenance | |
|---|---|---|---|---|
| aio_databases-2.0.0.tar.gz | 13.4 kB | Without provenance | Details |
Built distribution (wheel)
| File | Interpreter | ABI | Platform | Reset |
|---|---|---|---|---|
| aio_databases-2.0.0-py3-none-any.whl | Python 3 | none | any | Details |
Total release size: 32.1 kB
Release files / aio_databases-2.0.0.tar.gz
| Download URL | aio_databases-2.0.0.tar.gz |
|---|---|
| Size | 13.4 kB |
| Tags | Source |
|
SHA-256 checksum How to use checksums |
6aaa7f2a54c09e9208318e6ef71df9b5ce4a649a0c8eda117531a8153f4e2996
|
|
BLAKE2b-256 checksum How to use checksums |
6842c22d7e806aeb4060c7ccdc287df511e2afcf70d4c156b7dc47271400d860
|
| Upload date | |
|
Uploaded using Trusted Publishing? What is trusted publishing? |
No |
| Uploaded via |
uv/0.13.1 {"installer":{"name":"uv","version":"0.13.1","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-2.0.0-py3-none-any.whl
| Download URL | aio_databases-2.0.0-py3-none-any.whl |
|---|---|
| Size | 18.7 kB |
| Tags | Python 3 |
|
SHA-256 checksum How to use checksums |
40ef674d17875ec5d3464d45b48cfdb63a48ea004bc6e3be3941c00e6ec775e4
|
|
BLAKE2b-256 checksum How to use checksums |
ca27526aa7e3be4907f08128943718e7ee689544d32f49c6aef5537a71d84d11
|
| Upload date | |
|
Uploaded using Trusted Publishing? What is trusted publishing? |
No |
| Uploaded via |
uv/0.13.1 {"installer":{"name":"uv","version":"0.13.1","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}
|
Security for aio-databases 2.0.0
Provenance
Provenance describes where a file came from. On PyPI, provenance is shared via attestations, which provide a verifiable record of the build or publishing details. View details, limitations and caveats.
No provenance found
None of the files in this release were uploaded with provenance metadata.
A note on package security
PyPI is a repository for distributing packages, not a service for auditing them. While third-party security researchers and our staff do their best to detect and remove malicious packages, PyPI cannot guarantee the quality, security or suitability of any given package.
To learn more about protecting yourself from malicious actors, visit our package security documentation.
