sqlalchemy-aio 0.14.1

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(users.select(users.c.name.startswith('D')))
    d_users = await result.fetchall()

    await conn.close()

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

    # 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()
    loop.run_until_complete(main())

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)

    ...

trio.run(main)

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.

Documentation

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