Skip to main content

An async database toolkit.

Project description


A wrapper around SQLAlchemy made to support asynchronous workloads.

Aerie - is an avariel (or winged elf) from Baldur's Gate II game.

PyPI GitHub Workflow Status GitHub dependency status for latest release PyPI - Downloads GitHub Release Date Lines of code


Install aerie using PIP or poetry:

pip install aerie[postgresql]
# or
poetry add aerie[postgresql]

For SQLite use "sqlite" extra. To install all drivers use "full" extra.


  • full async/await support
  • plain SQL with bound params
  • SQLAlchemy query builders support
  • SQLAlchemy ORM support
  • pagination


  • simplify column definition: sa.Column(sa.Integer) -> models.IntergerField()
  • integrate with Alembic CLI

Quick start

See example application in examples/ directory of this repository.


As this library is based on SQLAlchemy, it is strictly recommended getting yourself familiar with it.

A general usage is:

  • create an instance of Aerie
  • define ORM models
  • create tables in the database (or, preferably, use Alembic migrations)
  • obtain a session and perform database queries

Aerie instance

Create an instance of Aerie class and pass a connection string to it:

from aerie import Aerie

db = Aerie('sqlite+aiosqlite:///tmp/database.sqlite2')
# or
db = Aerie('postgresql+asyncpg://postgres:postgres@localhost/aerie')

You need appropriate driver installed. Add "aiosqlite" for SQLite support, or add "asyncpg" for PostreSQL support.

Raw SQL queries

At this step Aerie is ready to work. Create a new transaction and execute any query you need.

from sqlalchemy.sql import text

# assuming "users" table exists
sql = text('select * from users where user_id = :user_id')
rows = await db.query(sql, {'user_id': 1}).all()

Full listing examples/

Using query builder

Sure, you are not limited to plain SQL. SQLAlchemy query builders also supported (because Aerie is a tiny layer on top of the SQLAlchemy)

from sqlalchemy.sql import text
import sqlalchemy as sa
from aerie import metadata

users = sa.Table(
    'users', metadata,
    sa.Column(sa.Integer, name='id', primary_key=True),
    sa.Column(sa.String, name='name'),

# create tables
await db.schema.create_tables()

stmt = select(users).where( == 2)
rows = await db.query(stmt).all()

Full listing examples/

Using ORM models and sessions

Another option to low-level query builder are ORM models. Aerie provides aerie.Model class that you should extend to create your model.

from aerie import Base
import sqlalchemy as sa

class User(Base):
    __tablename__ = 'users'

    id = sa.Column(sa.Integer, primary_key=True)
    name = sa.Column(sa.String)

# create tables
await db.schema.create_tables()

async with db.session() as session:
        User(id=1, name='One'),
        User(id=2, name='Two'),
        User(id=3, name='Three'),
    await session.flush()

    # get first user in the row set
    user = await session.query(User).first()

Make sure the module with models is imported before you create tables. Otherwise they will not be added to the metadata and, thus, not created.

Full listing examples/


Aerie's DbSession ships with pagination utilities out of the box. When you need to paginate a query just call DbSession.paginate method.

async with db.session() as session:
    page = await session.query(User).paginate(page=1, page_size=10)

    for user in page:

    print('Next page: %s' % page.next_page)
    print('Previous page: %s' % page.previous_page)
    print('Displaying items: %s - %s' % (page.start_index, page.end_index))

The page object has more helper attributes:

Property Type Description
total_pages int Total pages in the row set.
has_next bool Test if the next page is available.
has_previous bool Test if the previous page is available.
has_other bool Test if there are another pages except current one.
next_page int Next page number. Always returns an integer. If there is no more pages the current page number returned.
previous_page int Previous page number. Always returns an integer. If there is no previous page, the number 1 returned.
start_index int The 1-based index of the first item on this page.
end_index int The 1-based index of the last item on this page.
total_rows int Total rows in result set.

Alembic migrations

Alembic usage is well documented in the official docs: Using Asyncio with Alembic

Note, you need to use aerie.metadata when you configure target_metadata option:

# migrations/

from aerie import metadata

target_metadata = metadata

Also, don't forget to import all your models in Alembic's file so their structure is fully loaded and no models forgotten.

Shared instances

You can configure Aerie to populate Aerie.instances class-level variable, so you can access database instances from anywhere of your code. For that, just pass name argument to Aerie constructor.

# migrations/

from aerie import Aerie

db = Aerie(name='shared', ...)

# other file
db = Aerie.get_instance('shared')

Note, instances without name cannot be shared.

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

aerie-0.8.2.tar.gz (15.9 kB view hashes)

Uploaded source

Built Distribution

aerie-0.8.2-py3-none-any.whl (15.5 kB view hashes)

Uploaded py3

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