Adds support to asyncpgsa Record type for all the methods allowed in asyncpg Record type.

This change allows to use subscript operator and SQL Alchemy Core tables columns as indexes for the subscript.

On top of:

row.type_id
1

You can now:

row['type_id']
1
row[models.t_table.c.type_id]
1

It calls asyncpg Record for every other method call not implemented here, so you can use row.keys() or row.values() or row_list = [ dict(p) for p in rows ], the last being useful for serialization for example.

Related to #32

I have a table like the following:

CREATE TABLE users (
  id SERIAL,
  name VARCHAR(15),
  active DATERANGE,

  PRIMARY KEY (id)
)

that with plain SQLAlchemy+psycopg2 I can query with:

    today = date.today()
    q = select([users.c.id]) \
        .where(users.c.name == 'test') \
        .where(users.c.active.contains(today))
    r = session.execute(q)

With the following asyncpgsa transliteration:

    today = date.today()
    q = sa.select([users.c.id]) \
        .where(users.c.name == 'test') \
        .where(users.c.active.contains(today))
    r = await pg.fetchrow(q)

I get an error:

/usr/local/lib/python3.5/site-packages/asyncpgsa/pgsingleton.py:65: in fetchrow
    return await conn.fetchrow(query, *args, timeout=timeout)
/usr/local/lib/python3.5/site-packages/asyncpgsa/connection.py:91: in fetchrow
    result = await self.connection.fetchrow(query, *params, *args, **kwargs)
/usr/local/lib/python3.5/site-packages/asyncpg/connection.py:259: in fetchrow
    False, timeout)
asyncpg/protocol/protocol.pyx:157: in bind_execute (asyncpg/protocol/protocol.c:45856)
    ???
asyncpg/protocol/prepared_stmt.pyx:122: in asyncpg.protocol.protocol.PreparedStatementState._encode_bind_msg (asyncpg/protocol/protocol.c:42239)
    ???
asyncpg/protocol/codecs/base.pyx:123: in asyncpg.protocol.protocol.Codec.encode (asyncpg/protocol/protocol.c:12276)
    ???
asyncpg/protocol/codecs/base.pyx:90: in asyncpg.protocol.protocol.Codec.encode_range (asyncpg/protocol/protocol.c:11868)
    ???
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _

>   ???
E   TypeError: list, tuple or Range object expected (got type <class 'datetime.date'>)

asyncpg/protocol/codecs/range.pyx:53: TypeError

which seems odd to me: what I want is to test whether a single date is within a period...

I tried the following code, passing a one-item-tuple instead:

    today = date.today()
    q = sa.select([users.c.id]) \
        .where(users.c.name == 'test') \
        .where(users.c.active.contains((today,)))
    r = await pg.fetchrow(q)

that works, but that seems to convert the argument to a [today, +infinity] range, something different from what I'm seeking.

What am I missing?

We are using transaction as follows

async with pool.transaction() as conn:
    ...

and from time to time (in our use case it ranges from once a day to once a week) our service freezes. When we investigated the issue we noticed, that all connections in the pool were marked as used and no free connection was available. We traced the problem to the ConnectionTransactionContextManager.

If cancellation or timeout raises during __aenter__ or __aexit__ there is no guarantee that connection is returned to the pool. And it slowly drains. We confirmed that this is the issue, because we change code to this

async with pool.acquire() as conn:
    async with conn.transaction():
        ....

and our problem stops.

I can create PR to fix this if you like.

query = ...  # some SA select query
row = await conn.fetchrow(query)  # conn is SAConnection

When the underlying asyncpg connection returns None as the "empty" result, row should become None as well. Currently, row is not None but row.row is None. This would confuse the users who expect the API semantics to be same to asyncpg.

The same applies to fetch and execute which returns a RecordGenerator because it does not check None when instantiating Record objects.

Q: What's the purpose of having seem-to-be-redundant Record and RecordGenerator classes? They look like a no-op proxy to asyncpg's return objects.

Fixes #6

• Adds a debug param to various components to enable printing of some debug statements, currently only the query.
• Makes the connection protected to maintain library standard

I intentionally left off the params - I could see a case for someone wanting that at some point but there is often enough info in there you may not want shown.

As a start it just prints to stdout, not sure if you would like to use the logger for users to customize or print to stderr instead. In either case, I can happily fix it.

Hello!

I try to create table via this code:

import asyncio

from asyncpgsa import pg
import sqlalchemy as sa
from sqlalchemy.sql.ddl import CreateTable, DropTable

users = sa.Table(
    'users', sa.MetaData(),
    sa.Column('id', sa.Integer, primary_key=True),
    sa.Column('name', sa.VARCHAR(255)),
)


async def main():
    await pg.init('postgresql://localhost/test')
    await pg.query(DropTable(users))
    await pg.query(CreateTable(users))


loop = asyncio.get_event_loop()
loop.run_until_complete(main())

and got an error from asyncpgsa

Traceback (most recent call last):
  File "/project/main.py", line 20, in <module>
    loop.run_until_complete(main())
  File "/project/.venv/python3.6/asyncio/base_events.py", line 467, in run_until_complete
    return future.result()
  File "/project/main.py", line 15, in main
    await pg.query(DropTable(users))
  File "/project/.venv/lib/python3.6/site-packages/asyncpgsa/pgsingleton.py", line 65, in query
    compiled_q, compiled_args = compile_query(query)
  File "/project/.venv/lib/python3.6/site-packages/asyncpgsa/connection.py", line 60, in compile_query
    compiled_params = sorted(compiled.params.items())
AttributeError: 'NoneType' object has no attribute 'items'

Same code using sqlalchemy works fine:

import sqlalchemy as sa
from sqlalchemy.sql.ddl import CreateTable, DropTable

users = sa.Table(
    'users', sa.MetaData(),
    sa.Column('id', sa.Integer, primary_key=True),
    sa.Column('name', sa.VARCHAR(255)),
)

engine = sa.create_engine('postgresql://localhost/test')
engine.execute(DropTable(users))
engine.execute(CreateTable(users))

Thanks!

From version 0.14.0 conn.execute() is not working.

Simple example:

async with app.pool.acquire() as conn:
    query = table.update(table.c.some_culumn=123).where(table.c.id=1)
    await conn.execute(query)

Will go asyncpgsa/connection.py :

async def execute(self, script, *args, **kwargs) -> str:
        # script, params = compile_query(script, dialect=self._dialect)
        result = await super().execute(script, *args, **kwargs)
        return RecordGenerator(result)

And there *args will be empty tuple, hence when it would go to super().execute(script, *args, **kwargs) which is asyncpg.Connection.execute() :

async def execute(self, query: str, *args, timeout: float=None) -> str:
    ...
    if not args:  # <- WILL ENTER HERE
            return await self._protocol.query(query, timeout)

     _, status, _ = await self._execute(query, args, 0, timeout, True)
    return status.decode()

And if i will use version 0.13.0 everything will be alright, because i will pass *params and *args to asyncpg.Connection.execute() and then it will skip if not args clause in asyncpg.Connectuon.execute:

async def execute(self, script, *args, **kwargs) -> str:
    script, params = compile_query(script, dialect=self._dialect)
    result = await self._connection.execute(script, *params, *args, **kwargs)
    return RecordGenerator(result)

I am certainly grateful for that awesome package, and i am willing to help, but i not sure how, because recent change to omit compile_query in execute statement is made for a reason. And i don't really understand what reason it is.

For a moment i rolled back to 0.13.0 and everything works as expected.

This removes the Record and RecordGenerators in favor of the asyncpg Records. This means dot notation must be replaces with dict notation when accessing values, making this a breaking change.

It also fixes an issue where the _dialect property on the SAConnection was not getting set, causing issues when trying to use the query builder with table definitions containing JSON types. This is also potentially breaking, as it enables automatic type coercion into a JSON string, even if you have already passed the structure through json.dumps.

It also re-enables the use of compile_query for several of the SAConnection types, so that SA query-builder queries will get properly converted.

It also proxies the cursor method of the parent connection, allowing SA queries to be used with cursors directly.

Hi!

The new release of asyncpg 0.22.0 breaks compatibility with asyncpgsa and here is a simple fix for that. Please review and let me know if I need to fix something.

Thanks!

when i use pool and my db password is "dY8*6fN6Z#xSOg$wG9zDATTe" pool = await pg.create_pool(dsn) raise an error

hostlist_ports.append(int(hostspec_port))
ValueError: invalid literal for int() with base 10: 'dY8*6fN6Z'

When using asyncpgsa it's hard to use latest version of asyncpg, because there's asyncpg~=0.12.0 in install_requires. What's more, I'd like to use asyncpgsa only as sql query compiler, without using its context managers (as shown here). In this case asyncpg is not a asyncpgsa's dependency any more in practice. It's just a query compiler. How about moving asyncpg to extras_require? Or splitting the library into 2 separate packages (compiler and asyncpg adapter (any better name?)).

Today caught issue with incompatibility asyncpgsa==0.26.1 with asyncpg==0.22.0 (but found that latest release works well!).

So, i opened releases to see the diff between 0.26.1 and latest, but found the latest release described is 0.25. Perhaps, you would like to add other releases to make it easier to see the diffs?

Of course it is possible to build diff against the commits, but releases seem to be little bit faster/easier (and it is bit confusing that you have only part of releases on the github).

This works fine with asyncpgsa version 0.27.1 and sqlalchemy: 1.3.23, but fails when I upgrade to sqlalchemy 1.4.4.

import asyncpgsa
import sqlalchemy as sa

Thing = sa.Table("thing", sa.MetaData(), sa.Column("name", sa.Text))
query = Thing.insert().values(name="name").returning(Thing)
asyncpgsa.compile_query(query)

This prevents me from running any code that uses an INSERT statement.

The error seems to be here:

https://github.com/CanopyTax/asyncpgsa/blob/master/asyncpgsa/connection.py#L36

    if isinstance(query.parameters, list):

because sqlalchemy.sql.dml.Insert no longer has a parameters attribute.

For now I'm dealing with this by pinning to an earlier sqlalchemy version. It's not clear to me what the fix should be, because the sqlalchemy internals seem to have changed considerably between versions.

Please teach how to do Table(autoload=True) or metadata.reflect() with asyncpgsa properly.

Tried passing reflect(bind=asyncpgsa_connection) and it's AttributeError game: needs connect then dialect, I feel like it's going in wrong direction.

People use psycopg2 to create sqlalchemy engine and then async driver for normal work. That may work if reflection doesn't use driver specifics. I'll use that crutch if nothing else works.

Can you create sqlalchemy Engine with asyncpgsa?

I need TypeDecorator for my data column, it is handled by SQLAlchemy + psycopg2 correctly. Is it possible to make it work with asyncpgsa?

import asyncio
from datetime import datetime

from asyncpgsa import PG
from pytz import timezone
from sqlalchemy import (
    Column, DateTime, Integer, MetaData, Table, TypeDecorator, create_engine
)


DB_URL = 'postgresql://user:[email protected]/db'


class DateTime_(TypeDecorator):
    impl = DateTime

    def __init__(self):
        TypeDecorator.__init__(self, timezone=True)

    def process_bind_param(self, value, dialect):
        if value is not None:
            return datetime.fromtimestamp(value, timezone('UTC'))

    def process_result_value(self, value, dialect):
        return int(value.timestamp())


metadata = MetaData()
example_table = Table('example', metadata,
                      Column('id', Integer, primary_key=True),
                      Column('some_date', DateTime_))

engine = create_engine(DB_URL)

# Create table & add row
metadata.create_all(engine)
engine.execute(example_table.insert().values({
    'some_date': int(datetime.now().timestamp())
}))

# psycopg2 with sqlalchemy handles process_result_value correctly
rows = engine.execute(example_table.select()).fetchall()
assert isinstance(rows[0]['some_date'], int)


# asyncpgsa does not handle process_result_value callback
async def main():
    db = PG()
    await db.init(DB_URL)
    rows = await db.fetch(example_table.select())
    assert isinstance(rows[0]['some_date'], datetime)  # True
    assert isinstance(rows[0]['some_date'], int)  # False!


asyncio.run(main())

Perhaps such callbacks should be called in SAConnection.execute?

Perhaps you could advice some fast fix or workaround for that? I need update for many different rows with different values. Previously used bindparam for that.

import asyncio

from asyncpgsa import PG
from sqlalchemy import Table, MetaData, Column, Integer, String, bindparam, \
    create_engine

metadata = MetaData()

table = Table(
    'citizens',
    metadata,
    Column('test_id', Integer, primary_key=True),
    Column('name', String, nullable=False),
)

DB_URL = 'postgresql://user:[email protected]/db'


async def main():
    # create table
    engine = create_engine(DB_URL)
    metadata.create_all(engine)

    # connect to db
    pg = PG()
    await pg.init(DB_URL)
    async with pg.transaction() as conn:
        # create
        query = table.insert().values([
            {'name': str(i)} for i in range(10)
        ]).returning(table)
        rows = await conn.fetch(query)

        # update
        query = table.update().values(name=bindparam('name'))
        await conn.execute(query, [
            {'test_id': row['test_id'], 'name': row['name'] + '_new'}
            for row in rows
        ])

        # check
        # asyncpg.exceptions.NotNullViolationError: null value in column "name" violates not-null constraint
        # DETAIL:  Failing row contains (31, null).
        results = await conn.execute(table.select())
        print(results)

asyncio.run(main())