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

SQLAlchemy adaptor for asyncpg

Project Description

author:Lele Gaifax
contact:lele@metapensiero.it
license:GNU General Public License version 3 or later

This is a Python 3 package, spin-off from the proprietary Ytefas application, that implements the ability of executing SQLAlchemy core statements through asyncpg, in a cleaner way than asyncpgsa (I’m biased, of course 😜): the main difference is that it uses explicitly typed parameter placeholders, to avoid the problem that initially prompted me to write this package.

It emits DEBUG logs with prettified SQL statements, with parameters resolved to the actual values, and supplies an asyncpg variant of metapensiero.sqlalchemy.proxy ‘s ProxiedQuery.

Typical usage

The following script:

import asyncio
from datetime import date
import logging

import sqlalchemy as sa
from asyncpg import create_pool
from asyncpg.types import Range
from metapensiero.sqlalchemy import asyncpg as sasyncpg


async def dml_tests(connection):
    sasyncpg_test = sa.Table('sasyncpg_test', sa.MetaData(),
                             sa.Column('id', sa.types.Integer, primary_key=True),
                             sa.Column('value', sa.types.Text),
                             sa.Column('period', sa.dialects.postgresql.DATERANGE))

    value = 'First test'
    insert_stmt = sasyncpg_test.insert().values(id=1, value=value)
    await connection.execute(insert_stmt)

    new_value = 'Second test'
    update_stmt = (sasyncpg_test.update()
                   .values(value=new_value,
                           period=Range(date(2016, 2, 1), date(2016, 3, 1)))
                   .where(sasyncpg_test.c.id == 1))
    await connection.execute(update_stmt)

    select_stmt = (sa.select([sasyncpg_test.c.value])
                   .where(sasyncpg_test.c.id == sa.bindparam('id')))
    for row in await connection.fetchall(select_stmt, named_args={'id': 1}):
        print('Row:', row)

    single_row = (sa.select([sasyncpg_test])
                  .where(sasyncpg_test.c.period.contains(date(2016, 2, 15))))
    print('Row:', await connection.fetchone(single_row))


async def run(loop):
    pool = await create_pool(database="test", loop=loop)

    async with pool.acquire() as apgc:
        connection = sasyncpg.Connection(apgc)

        query = sa.select([sa.func.version()])
        result = await connection.scalar(query)
        print("PostgreSQL version:", result)

        await connection.execute('create table sasyncpg_test ('
                                 ' id integer not null primary key,'
                                 ' value text,'
                                 ' period daterange)')

        try:
            await dml_tests(connection)
        finally:
            await connection.execute('DROP TABLE sasyncpg_test')


def main():
    loop = asyncio.get_event_loop()
    loop.run_until_complete(run(loop))


if __name__ == '__main__':
    logging.basicConfig(level=logging.DEBUG)
    main()

produces something like:

DEBUG:asyncio:Using selector: EpollSelector
DEBUG:metapensiero.sqlalchemy.asyncpg.funcs:Fetching scalar in transaction 9ddb60:
    SELECT version() AS version_1
DEBUG:metapensiero.sqlalchemy.asyncpg.funcs:Fetched value in 869 µsec
PostgreSQL version: PostgreSQL 9.6.6 on x86_64-pc-linux-gnu, compiled by gcc (Debian 7.2.0-12) 7.2.1 20171025, 64-bit
DEBUG:metapensiero.sqlalchemy.asyncpg.funcs:Executing in transaction 9ddb60:
    CREATE TABLE sasyncpg_test (
        id integer NOT NULL PRIMARY KEY, value text, period daterange
    )
DEBUG:metapensiero.sqlalchemy.asyncpg.funcs:Execution took 97.1 msec
DEBUG:metapensiero.sqlalchemy.asyncpg.funcs:Executing in transaction 9ddb60:
    INSERT INTO sasyncpg_test (id, value)
    VALUES (1::integer, 'First test'::text)
DEBUG:metapensiero.sqlalchemy.asyncpg.funcs:Execution took 1.26 msec
DEBUG:metapensiero.sqlalchemy.asyncpg.funcs:Executing in transaction 9ddb60:
    UPDATE sasyncpg_test
    SET value = 'Second test'::text, period = '[2016-02-01,2016-03-01)'::daterange
    WHERE sasyncpg_test.id = 1::integer
DEBUG:metapensiero.sqlalchemy.asyncpg.funcs:Execution took 14.1 msec
DEBUG:metapensiero.sqlalchemy.asyncpg.funcs:Fetching rows in transaction 9ddb60:
    SELECT sasyncpg_test.value
    FROM sasyncpg_test
    WHERE sasyncpg_test.id = 1::integer
DEBUG:metapensiero.sqlalchemy.asyncpg.funcs:Fetched 1 records in 909 µsec
Row: <Record value='Second test'>
DEBUG:metapensiero.sqlalchemy.asyncpg.funcs:Fetching row in transaction 9ddb60:
    SELECT sasyncpg_test.id, sasyncpg_test.value, sasyncpg_test.period
    FROM sasyncpg_test
    WHERE sasyncpg_test.period @> datetime.date(2016, 2, 15)::date
DEBUG:metapensiero.sqlalchemy.asyncpg.funcs:Fetched one record in 951 µsec
Row: <Record id=1 value='Second test' period=<Range [datetime.date(2016, 2, 1), datetime.date(2016, 3, 1))>>
DEBUG:metapensiero.sqlalchemy.asyncpg.funcs:Executing in transaction 9ddb60:
    DROP TABLE sasyncpg_test RESTRICT
DEBUG:metapensiero.sqlalchemy.asyncpg.funcs:Execution took 12 msec

Changes

0.1 (2017-12-03)

  • Standalone package, released under GPLv3

0.0 (unreleased)

  • Initial effort.

Release History

This version
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.

File Name & Hash SHA256 Hash Help Version File Type Upload Date
metapensiero.sqlalchemy.asyncpg-0.1.tar.gz
(19.6 kB) Copy SHA256 Hash SHA256
Source Dec 3, 2017

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