Async support for SQLAlchemy.

async, async-await, asyncio, python, sqlalchemy
pip install sqlalchemy-aio==0.17.0



PyPI Version Documentation Travis Coverage MIT License

sqlalchemy_aio adds asyncio and Trio support to SQLAlchemy core, derived from alchimia.

Getting started

import asyncio

from sqlalchemy_aio import ASYNCIO_STRATEGY

from sqlalchemy import (
    Column, Integer, MetaData, Table, Text, create_engine, select)
from sqlalchemy.schema import CreateTable, DropTable

async def main():
    engine = create_engine(
        # In-memory sqlite database cannot be accessed from different
        # threads, use file.
        'sqlite:///test.db', strategy=ASYNCIO_STRATEGY

    metadata = MetaData()
    users = Table(
        'users', metadata,
        Column('id', Integer, primary_key=True),
        Column('name', Text),

    # Create the table
    await engine.execute(CreateTable(users))

    conn = await engine.connect()

    # Insert some users
    await conn.execute(users.insert().values(name='Jeremy Goodwin'))
    await conn.execute(users.insert().values(name='Natalie Hurley'))
    await conn.execute(users.insert().values(name='Dan Rydell'))
    await conn.execute(users.insert().values(name='Casey McCall'))
    await conn.execute(users.insert().values(name='Dana Whitaker'))

    result = await conn.execute('D')))
    d_users = await result.fetchall()

    await conn.close()

    # Print out the users
    for user in d_users:
        print('Username: %s' % user[])

    # Supports context async managers
    async with engine.connect() as conn:
        async with conn.begin() as trans:
            assert await conn.scalar(select([1])) == 1

    await engine.execute(DropTable(users))

if __name__ == '__main__':
    loop = asyncio.get_event_loop()

Getting started with Trio

To use the above example with Trio, just change the following:

import trio
from sqlalchemy_aio import TRIO_STRATEGY

async def main():
    engine = create_engine('sqlite:///test.db', strategy=TRIO_STRATEGY)


What is this?

It's not an asyncio implementation of SQLAlchemy or the drivers it uses. sqlalchemy_aio lets you use SQLAlchemy by running operations in a separate thread.

If you're already using run_in_executor to execute SQLAlchemy tasks, sqlalchemy_aio will work well with similar performance. If performance is critical, perhaps asyncpg can help.


The documentation has more information, including limitations of the API.