metapensiero.sqlalchemy.asyncio 1.0

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

Changes