Skip to main content
This is a pre-production deployment of Warehouse. Changes made here affect the production instance of PyPI (
Help us improve Python packaging - Donate today!

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 arstecnica.sqlalchemy.async 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 arstecnica.sqlalchemy.async 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 and either psycopg2 or pymysql:

    pip install pytest psycopg2 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 src
    TEST_DB_URL="mysql+pymysql://localhost/testdb" py.test src


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.

Release History

Release History

This version
History Node


History Node


History Node


History Node


Download Files

Download Files

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

File Name & Checksum SHA256 Checksum Help Version File Type Upload Date
arstecnica.sqlalchemy.async-0.4.tar.gz (6.3 kB) Copy SHA256 Checksum SHA256 Source Sep 25, 2015

Supported By

WebFaction WebFaction Technical Writing Elastic Elastic Search Pingdom Pingdom Monitoring Dyn Dyn DNS Sentry Sentry Error Logging CloudAMQP CloudAMQP RabbitMQ Heroku Heroku PaaS Kabu Creative Kabu Creative UX & Design Fastly Fastly CDN DigiCert DigiCert EV Certificate Rackspace Rackspace Cloud Servers DreamHost DreamHost Log Hosting