metapensiero.sqlalchemy.asyncio

A port of Alchimia to Python 3.4 asyncio

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.

Usage

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

@coroutine
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(users.select()
                                      .where(users.c.id == 42))
        rows = yield from res.fetchall()

        res = yield from conn.execute(users.delete()
                                      .where(users.c.id == 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(users.select()
                                 .where(users.c.id == 42))
        rows = await res.fetchall()

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

Tests

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 setup.py 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