Skip to main content

Databases + asyncio.

Project description


Build Status Coverage Package version

Databases gives you simple asyncio support for a range of databases.

Currently PostgreSQL and MySQL are supported.

Requirements: Python 3.6+


$ pip install databases

You can install the required database drivers with:

$ pip install databases[postgresql]
$ pip install databases[mysql]

Getting started

Declare your tables using SQLAlchemy:

import sqlalchemy

metadata = sqlalchemy.MetaData()

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

You can use any of the sqlalchemy column types such as sqlalchemy.JSON, or custom column types.


You can now use any SQLAlchemy core queries:

from databases import Database

database = Database('postgresql://localhost/example')

# Establish the connection pool
await database.connect()

# Execute
query = notes.insert()
values = {"text": "example1", "completed": True}
await database.execute(query, values)

# Execute many
query = notes.insert()
values = [
    {"text": "example2", "completed": False},
    {"text": "example3", "completed": True},
await database.execute_many(query, values)

# Fetch multiple rows
query =
rows = await database.fetch_all(query)

# Fetch single row
query =
row = await database.fetch_one(query)

# Fetch multiple rows without loading them all into memory at once
query =
async for row in database.iterate(query):


Transactions are managed by async context blocks:

async with database.transaction():

For a lower-level transaction API:

transaction = await database.transaction()

You can also use .transaction() as a function decorator on any async function:

async def create_users(request):

Transaction blocks are managed as task-local state. Nested transactions are fully supported, and are implemented using database savepoints.

Connecting and disconnecting

You can control the database connect/disconnect, by using it as a async context manager.

async with Database(DATABASE_URL) as database:

Or by using explicit connection and disconnection:

database = Database(DATABASE_URL)
await database.connect()
await database.disconnect()

Test isolation

For strict test isolation you will always want to rollback the test database to a clean state between each test case:

database = Database(DATABASE_URL, force_rollback=True)

This will ensure that all database connections are run within a transaction that rollbacks once the database is disconnected.

If you're integrating against a web framework you'll typically want to use something like the following pattern:

if not TESTING:
    database = Database(DATABASE_URL)
    database = Database(TEST_DATABASE_URL, force_rollback=True)

This will give you test cases that run against a different database to the development database, with strict test isolation so long as you make sure to connect and disconnect to the database between test cases.

For a lower level API you can explicitly create force-rollback transactions:

async with database.transaction(force_rollback=True):

Project details

Download files

Download the file for your platform. If you're not sure which to choose, learn more about installing packages.

Source Distribution

databases-0.1.2.tar.gz (11.1 kB view hashes)

Uploaded Source

Supported by

AWS AWS Cloud computing and Security Sponsor Datadog Datadog Monitoring Fastly Fastly CDN Google Google Download Analytics Microsoft Microsoft PSF Sponsor Pingdom Pingdom Monitoring Sentry Sentry Error logging StatusPage StatusPage Status page