Databases

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

Currently PostgreSQL and MySQL are supported.

Requirements: Python 3.6+

---

Installation

$ 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(
    "notes",
    metadata,
    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.

Queries

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 = notes.select()
rows = await database.fetch_all(query)


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

Transactions

Transactions are managed by async context blocks:

async with database.transaction():
    ...

For a lower-level transaction API:

transaction = await database.transaction()
try:
    ...
except:
    transaction.rollback()
else:
    transaction.commit()

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

async with database.transaction(force_rollback=True):
    ...

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