Skip to main content

Asyncio middleware for SQLAlchemy

Project description

The aiopg and aiomysql projects allow to operate asynchronously with PostgresQL and MySQL, respectively, even thru SQLAlchemy. Their approach isn’t ideal, though, because they reimplement a considerable amount of SA low level stuff with undesirable glitches.

The Twisted based Alchimia way is much lighter and even if maybe slightly less performant it does not introduce unexpected surprises.


Basically the module wraps a minimal set of SA classes (currently just Engine, Connection, Transaction and ResultProxy) into asyncronous counterparts, and you work on them like the following example:

from asyncio import coroutine
from metapensiero.sqlalchemy.asyncio import create_engine

def do_something(db_url):
    engine = create_engine(db_url)
    with (yield from engine.connect()) as conn:
        with (yield from conn.begin()) as trans:
            yield from conn.execute(users.insert()
                                    .values(id=42, name="Async",))

        res = yield from conn.execute(
                                      .where( == 42))
        rows = yield from res.fetchall()

        res = yield from conn.execute(users.delete()
                                      .where( == 42))
        assert res.rowcount == 1

If you are using Python 3.5 or better, the above should be written as:

from metapensiero.sqlalchemy.asyncio import create_engine

async def do_something(db_url):
    engine = create_engine(db_url)
    async with await engine.connect() as conn:
        async with await conn.begin() as trans:
            await conn.execute(users.insert()
                               .values(id=42, name="Async",))

        res = await conn.execute(
                                 .where( == 42))
        rows = await res.fetchall()

        res = await conn.execute(users.delete()
                                 .where( == 42))
        assert res.rowcount == 1


To run the unit tests, you should:

  1. create a Python virtual environment and install this package in development mode:

    python3 -m venv env
    source env/bin/activate
    python develop
  2. install pytest, pytest-asyncio and either psycopg2-binary or pymysql:

    pip install pytest pytest-asyncio psycopg2-binary pymysql
  3. create a test database, for example with createdb testdb

  4. execute the py.test runner with an environment variable with the SA URL of the db:

    TEST_DB_URL="postgresql://localhost/testdb" py.test tests
    TEST_DB_URL="mysql+pymysql://localhost/testdb" py.test tests


1.0 (2018-07-01)

  • Renamed to metapensiero.sqlalchemy.asyncio

0.4 (2015-09-25)

  • Packaging tweaks

0.3 (2015-09-23)

  • Support Python 3.5 asynchronous context managers

0.2 (2015-09-09)

  • First (usable) distribution on PyPI

0.1 (private)

Works reasonably well!

0.0 (private)

Initial effort.

Project details

Release history Release notifications

This version
History Node


Download files

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

Filename, size & hash SHA256 hash help File type Python version Upload date
metapensiero.sqlalchemy.asyncio-1.0.tar.gz (6.4 kB) Copy SHA256 hash SHA256 Source None

Supported by

Elastic Elastic Search Pingdom Pingdom Monitoring Google Google BigQuery Sentry Sentry Error logging AWS AWS Cloud computing DataDog DataDog Monitoring Fastly Fastly CDN SignalFx SignalFx Supporter DigiCert DigiCert EV certificate StatusPage StatusPage Status page