SQLAlchemy adaptor for asyncpg
Project description
- author:
Lele Gaifax
- contact:
- 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.
Project details
Release history Release notifications | RSS feed
Download files
Download the file for your platform. If you're not sure which to choose, learn more about installing packages.
Source Distribution
File details
Details for the file metapensiero.sqlalchemy.asyncpg-0.1.tar.gz
.
File metadata
- Download URL: metapensiero.sqlalchemy.asyncpg-0.1.tar.gz
- Upload date:
- Size: 19.6 kB
- Tags: Source
- Uploaded using Trusted Publishing? No
File hashes
Algorithm | Hash digest | |
---|---|---|
SHA256 | 0399642907cb00d3c9f2c8401867747defd3f19981d4af8b9a9153c0d4599dcb |
|
MD5 | b2b583094f60f78fffcb634093148a71 |
|
BLAKE2b-256 | dab1c94df67930b6fe2e382734f991a1908207166dbf2f322ff92f06a777c2e9 |