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.

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 arstecnica.sqlalchemy.async 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 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(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 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
    

Changes

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

0.4

History Node

0.3

History Node

0.2

History Node

0.1

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
arstecnica.sqlalchemy.async-0.4.tar.gz (6.3 kB) Copy SHA256 hash SHA256 Source None Sep 25, 2015

Supported by

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