aiopg 1.1.0+ is incompatible with SQLAlchemy 1.4

[Velikolay]

Aiopg will automatically download SQLAlchemy 1.4.0(Released March 15) which leads to crazy errors due to incompatibility. Make sure to freeze SQLAlchemy==1.3.23.

Two specific issues i noticed:

1. Failure when doing a select. What i found that the query is absolutely valid and Postgres returns what is expected, but the RowProxy fails to map the columns.

   
   self = <[InvalidRequestError("Ambiguous column name 'None' in result set! try 'use_labels' option on select statement.") raised in repr()] RowProxy object at 0x7fccab231460>
   key = 'brands_id'
   
    def __getitem__(self, key):
        try:
   >           processor, obj, index = self._keymap[key]
   E           KeyError: 'brands_id'

/usr/local/lib/python3.7/site-packages/aiopg/sa/result.py:27: KeyError

During handling of the above exception, another exception occurred:

tables = None, sa_engine = <aiopg.sa.engine.Engine object at 0x7fccab2813d0>

2. Failure when Deleting from table. Honestly i'm not sure if aiopg is responsible for that, it may very well be psycopg2.

E psycopg2.errors.SyntaxError: syntax error at or near "[" E LINE 1: ...ETE FROM materials WHERE materials.product_id IN ([POSTCOMPI... E ^

/usr/local/lib/python3.7/site-packages/aiopg/connection.py:106: SyntaxError

[Arfey]

Hi ✋ Could you explain a little more about the crazy incompatibility errors? maybe some example or something like that

[Velikolay]

@Arfey excuse my overly simplified explanation - provided more details in an edit. I hope it helps.

[krkd]

2. Second happens due to adding caching layer to sqlalchemy 1.4.2 https://docs.sqlalchemy.org/en/14/changelog/migration_14.html#all-in-expressions-render-parameters-for-each-value-in-the-list-on-the-fly-e-g-expanding-parameters

In order to use cache effectively literal values are precompiled to [POSTCOMPILE_\d+] and substituted before execution, after cache lookup

[Pliner]

@krkd thanks. Do you have any ideas how to fix it?

[krkd]

Not at the moment. Need to get familiar with changes introduced in new SQLAlchemy release

[Pliner]

@krkd feel free to raise a PR 😉

[Pliner]

@Velikolay @Arfey @krkd Could you try latest beta please(with fix from @AVOstap, 1.3.0b1) and confirm that it works for you?

[WouldYouKindly]

@Pliner Hi! I get the same error on 1.3.0b3.

async with self.engine.acquire() as conn:
    async with conn.execute(query, params) as proxy:
        rows = await proxy.fetchall()
        rows[0].items()

Gets me

    def __getitem__(self, key):
        try:
            processor, obj, index = self._keymap[key]
        except KeyError:
            processor, obj, index = self._result_proxy._key_fallback(key)
        # Do we need slicing at all? RowProxy now is Mapping not Sequence
        # except TypeError:
        #     if isinstance(key, slice):
        #         l = []
        #         for processor, value in zip(self._processors[key],
        #                                     self._row[key]):
        #             if processor is None:
        #                 l.append(value)
        #             else:
        #                 l.append(processor(value))
        #         return tuple(l)
        #     else:
        #         raise
        if index is None:
>           raise exc.InvalidRequestError(
                f"Ambiguous column name {key!r} in result set! "
                f"try 'use_labels' option on select statement."
            )
E           aiopg.sa.exc.InvalidRequestError: Ambiguous column name None in result set! try 'use_labels' option on select statement.

[Pliner]

@Pliner Hi! I get the same error on 1.3.0b3.

async with self.engine.acquire() as conn:
    async with conn.execute(query, params) as proxy:
        rows = await proxy.fetchall()
        rows[0].items()

Gets me

    def __getitem__(self, key):
        try:
            processor, obj, index = self._keymap[key]
        except KeyError:
            processor, obj, index = self._result_proxy._key_fallback(key)
        # Do we need slicing at all? RowProxy now is Mapping not Sequence
        # except TypeError:
        #     if isinstance(key, slice):
        #         l = []
        #         for processor, value in zip(self._processors[key],
        #                                     self._row[key]):
        #             if processor is None:
        #                 l.append(value)
        #             else:
        #                 l.append(processor(value))
        #         return tuple(l)
        #     else:
        #         raise
        if index is None:
>           raise exc.InvalidRequestError(
                f"Ambiguous column name {key!r} in result set! "
                f"try 'use_labels' option on select statement."
            )
E           aiopg.sa.exc.InvalidRequestError: Ambiguous column name None in result set! try 'use_labels' option on select statement.

It should be fixed by #870.

[WouldYouKindly]

@Pliner Hi, I don't think it is fixed. The PR fixes an issue with string_or_unprintable, but doesn't fix the my example. I've just tested it on aiopg 1.3.2b1 and sqlalchemy 1.4.20, and the issue persists.

[Pliner]

@WouldYouKindly Sad news.

Could you create a minimal example that reproduces the issue, please?

I've updated a lot of services in our company and 1.3.1 works fine. It looks like we use aiopg+sqlalchemy in a bit different way.

[WouldYouKindly]

@Pliner it seems that apply_labels is the culprit. We have to use it when joining two tables, as they have the same id column. The following example doesn't throw any errors if I remove apply_labels.

import pytest
import sqlalchemy
from aiopg.sa import InvalidRequestError
from sqlalchemy import select, Column, Integer, String
from sqlalchemy.orm import declarative_base
from sqlalchemy.pool import NullPool

Base = declarative_base()

class Model(Base):
      __tablename__ = "sa_tbl"

      id = Column(Integer, primary_key=True)
      code = Column(String(50))

@pytest.mark.asyncio
async def test_apply_labels():
    engine = sqlalchemy.create_engine(dsn)

    async with engine.acquire() as conn:
        await conn.execute(
            "CREATE TABLE sa_tbl (id serial primary key, code text)"
        )

        await conn.execute(
            "insert into sa_tbl (code) values ('a')"
        )

        query = select(Model).apply_labels()

        async with conn.execute(query) as proxy:
            rows = await proxy.fetchall()
            with pytest.raises(InvalidRequestError):
                dict(rows[0])

            with pytest.raises(InvalidRequestError):
                for _, __ in rows[0].items():
                    pass

            with pytest.raises(AttributeError):
                rows[0].sa_tbl_code

(You have to provide your own dsn, but other than that the example should work)

[decaz]

I have the same error. Simple snippet to reproduce it:

import asyncio
import logging

import sqlalchemy as sa
from aiopg.sa import create_engine

logging.basicConfig(level=logging.INFO)

metadata = sa.MetaData()
tbl = sa.Table('tbl', metadata,
               sa.Column('id', sa.Integer, primary_key=True),
               sa.Column('val', sa.String(255)))

dsn = ''

async def main():
    async with create_engine(dsn, echo=True) as engine:
        async with engine.acquire() as conn:
            await conn.execute('DROP TABLE IF EXISTS tbl')
            await conn.execute('''CREATE TABLE tbl (
                                 id serial PRIMARY KEY,
                                 val varchar(255))''')
            await conn.execute('INSERT INTO tbl (val) VALUES (\'test1\'), (\'test2\')')
            query = sa.select((tbl,), use_labels=True)
            async for row in conn.execute(query):
                print(row)

asyncio.run(main())

Running with SQLAlchemy 1.3.24:

INFO:aiopg:SELECT t.oid, typarray FROM pg_type t JOIN pg_namespace ns ON typnamespace = ns.oid WHERE typname = 'hstore';
INFO:aiopg:None
INFO:aiopg:SELECT tbl.id AS tbl_id, tbl.val AS tbl_val 
FROM tbl
INFO:aiopg:{}
(1, 'test1')
(2, 'test2')

Running with SQLAlchemy 1.4.20:

INFO:aiopg:SELECT t.oid, typarray FROM pg_type t JOIN pg_namespace ns ON typnamespace = ns.oid WHERE typname = 'hstore';
INFO:aiopg:None
INFO:aiopg:SELECT tbl.id AS tbl_id, tbl.val AS tbl_val 
FROM tbl
INFO:aiopg:{}
Traceback (most recent call last):
  File "/home/decaz/workspace/xyz/xyz.py", line 32, in <module>
    asyncio.run(main())
  File "/home/decaz/.pyenv/versions/3.9.6/lib/python3.9/asyncio/runners.py", line 44, in run
    return loop.run_until_complete(main)
  File "/home/decaz/.pyenv/versions/3.9.6/lib/python3.9/asyncio/base_events.py", line 642, in run_until_complete
    return future.result()
  File "/home/decaz/workspace/xyz/xyz.py", line 29, in main
    print(row)
  File "/home/decaz/workspace/xyz/.venv/lib/python3.9/site-packages/aiopg/sa/result.py", line 79, in __repr__
    return repr(self.as_tuple())
  File "/home/decaz/workspace/xyz/.venv/lib/python3.9/site-packages/aiopg/sa/result.py", line 76, in as_tuple
    return tuple(self[k] for k in self)
  File "/home/decaz/workspace/xyz/.venv/lib/python3.9/site-packages/aiopg/sa/result.py", line 76, in <genexpr>
    return tuple(self[k] for k in self)
  File "/home/decaz/workspace/xyz/.venv/lib/python3.9/site-packages/aiopg/sa/result.py", line 44, in __getitem__
    raise exc.InvalidRequestError(
aiopg.sa.exc.InvalidRequestError: Ambiguous column name None in result set! try 'use_labels' option on select statement.

[decaz]

@Pliner after some research I guess I've found the root of the error: https://github.com/aio-libs/aiopg/blob/7e4c8adee6e4d6fc07a847f5267503ba9e18281d/aiopg/sa/result.py#L174 elem[2][0] is an instance of sqlalchemy.sql.compiler._CompileLabel which in SQLAlchemy 1.3 was inherited from sqlalchemy.sql.visitors.Visitable but in version 1.4 it is inherited from sqlalchemy.sql.elements.ColumnElement which is always has the key attribute and in current case its value is None so the value of priority_name will be None also.

Maybe @zzzeek can help us here with some workaround to make this work with SQLAlchemy 1.4, please?

[zzzeek]

it looks like you're peeking in the result map, the format of this map has changed but still has the same information inside of it so just pdb and take a look how to get the info you need.

bigger picture, I've always thought it was a pretty bad idea to copy internal source code from SQLAlchemy and then try to keep it in sync. The way statement execution and results work in 1.4 is completely changed from 1.3, IMO it is not really worth the effort to keep chasing SQLAlchemy internals. aiopg should work on ripping all of that reimplementation out and just creating the async API they want using the approach our own asyncio adapter now uses. You will have vastly less code to deal with, little to no need to track our internal implementations and vastly more future proof.

[serg666]

Hi, guys!

Have the same error as @decaz

If I set param use_labels to True then the KeyError has appear

import logging
import asyncio
from aiopg.sa import create_engine
import sqlalchemy as sa

logging.basicConfig(level=logging.INFO)

metadata = sa.MetaData()

tbl = sa.Table('t1', metadata,
               sa.Column('id', sa.Integer, primary_key=True),
               sa.Column('val', sa.String(255)))

async def go():
    async with create_engine(user='aiopg',
                             database='aiopg',
                             host='127.0.0.1',
                             password='aiopg', echo=True) as engine:
        async with engine.acquire() as conn:
            await conn.execute(tbl.insert().values(val='abc'))

            async for row in conn.execute(tbl.select(use_labels=True).where(tbl.c.val=='abc')):
                print(row.t1_id, row.t1_val)

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

runing with SQLAlchemy-1.4.25 and aiopg-1.3.1 I have got the following:

INFO:aiopg:SELECT t.oid, typarray FROM pg_type t JOIN pg_namespace ns ON typnamespace = ns.oid WHERE typname = 'hstore';
INFO:aiopg:None
INFO:aiopg:INSERT INTO t1 (val) VALUES (%(val)s) RETURNING t1.id
INFO:aiopg:{'val': 'abc'}
INFO:aiopg:SELECT t1.id AS t1_id, t1.val AS t1_val 
FROM t1 
WHERE t1.val = %(val_1)s
INFO:aiopg:{'val_1': 'abc'}
Traceback (most recent call last):
  File "/home/mamonov/env-gateway/lib64/python3.8/site-packages/aiopg/sa/result.py", line 27, in __getitem__
    processor, obj, index = self._keymap[key]
KeyError: 't1_id'

During handling of the above exception, another exception occurred:

Traceback (most recent call last):
  File "test.py", line 27, in <module>
    loop.run_until_complete(go())
  File "/usr/lib64/python3.8/asyncio/base_events.py", line 616, in run_until_complete
    return future.result()
  File "test.py", line 23, in go
    print(row.t1_id, row.t1_val)
  File "/home/mamonov/env-gateway/lib64/python3.8/site-packages/aiopg/sa/result.py", line 55, in __getattr__
    return self[name]
  File "/home/mamonov/env-gateway/lib64/python3.8/site-packages/aiopg/sa/result.py", line 29, in __getitem__
    processor, obj, index = self._result_proxy._key_fallback(key)
  File "/home/mamonov/env-gateway/lib64/python3.8/site-packages/aiopg/sa/result.py", line 200, in _key_fallback
    f"Could not locate column in row for column "
AttributeError: module 'sqlalchemy.sql.expression' has no attribute '_string_or_unprintable'

If I set param use_labels to False then no error has appear

import logging
import asyncio
from aiopg.sa import create_engine
import sqlalchemy as sa

logging.basicConfig(level=logging.INFO)

metadata = sa.MetaData()

tbl = sa.Table('t1', metadata,
               sa.Column('id', sa.Integer, primary_key=True),
               sa.Column('val', sa.String(255)))

async def go():
    async with create_engine(user='aiopg',
                             database='aiopg',
                             host='127.0.0.1',
                             password='aiopg', echo=True) as engine:
        async with engine.acquire() as conn:
            await conn.execute(tbl.insert().values(val='abc'))

            async for row in conn.execute(tbl.select(use_labels=False).where(tbl.c.val=='abc')):
                print(row.id, row.val)

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

runing with SQLAlchemy-1.4.25 and aiopg-1.3.1 I have got the following:

INFO:aiopg:SELECT t.oid, typarray FROM pg_type t JOIN pg_namespace ns ON typnamespace = ns.oid WHERE typname = 'hstore';
INFO:aiopg:None
INFO:aiopg:INSERT INTO t1 (val) VALUES (%(val)s) RETURNING t1.id
INFO:aiopg:{'val': 'abc'}
INFO:aiopg:SELECT t1.id, t1.val 
FROM t1 
WHERE t1.val = %(val_1)s
INFO:aiopg:{'val_1': 'abc'}
1 abc
2 abc
3 abc
4 abc
5 abc
6 abc
7 abc
8 abc
9 abc
10 abc
11 abc
12 abc
13 abc
14 abc
15 abc
16 abc
17 abc
18 abc
19 abc
20 abc
21 abc
22 abc
23 abc
24 abc

Any ideas, when this will be fixed?

[asvetlov]

The proper fix is dropping aiopg.sa subsystem at all.

You can 'apply' this fix right now by just switching from aiopg.sa usage to sqlalchemy in async mode. Years ago, when sqlalchemy didn't support async, aiopg.sa made a value. Now it is just garbage, please use upstream.

[serg666]

It requires a lot of code rewriting, i think )

[serg666]

You can 'apply' this fix right now by just switching from aiopg.sa usage to sqlalchemy in async mode.

Or by switching from python at all)

[Pliner]

Anyway, here is a new version https://pypi.org/project/aiopg/1.3.2b2/ with a fix from @AVOstap.

@serg666 Could you test it please?

[serg666]

Hello, @Pliner ! Sure!

pip install aiopg==1.3.2b2
Collecting aiopg==1.3.2b2
  Downloading https://files.pythonhosted.org/packages/4a/f1/1b9a64e90dae1c3ede539d9e665c26f0b721f0fb1ad84ef85fc5b9342126/aiopg-1.3.2b2-py3-none-any.whl
Requirement already satisfied: async-timeout<4.0,>=3.0 in ./lib/python3.8/site-packages (from aiopg==1.3.2b2) (3.0.1)
Requirement already satisfied: psycopg2-binary>=2.8.4 in ./lib/python3.8/site-packages (from aiopg==1.3.2b2) (2.8.5)
Installing collected packages: aiopg
  Found existing installation: aiopg 1.3.1
    Uninstalling aiopg-1.3.1:
      Successfully uninstalled aiopg-1.3.1
Successfully installed aiopg-1.3.2b2

with the code below

import logging
import asyncio
from aiopg.sa import create_engine
import sqlalchemy as sa

logging.basicConfig(level=logging.INFO)

metadata = sa.MetaData()

tbl = sa.Table('t1', metadata,
               sa.Column('id', sa.Integer, primary_key=True),
               sa.Column('val', sa.String(255)))

async def go():
    async with create_engine(user='aiopg',
                             database='aiopg',
                             host='127.0.0.1',
                             password='aiopg', echo=True) as engine:
        async with engine.acquire() as conn:
            await conn.execute(tbl.insert().values(val='abc'))

            async for row in conn.execute(tbl.select(use_labels=True).where(tbl.c.val=='abc')):
                print(row.t1_id, row.t1_val)

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

running with SQLAlchemy-1.4.25 and aiopg-1.3.2b2 there is no problem

INFO:aiopg:SELECT t.oid, typarray FROM pg_type t JOIN pg_namespace ns ON typnamespace = ns.oid WHERE typname = 'hstore';
INFO:aiopg:None
INFO:aiopg:INSERT INTO t1 (val) VALUES (%(val)s) RETURNING t1.id
INFO:aiopg:{'val': 'abc'}
INFO:aiopg:SELECT t1.id AS t1_id, t1.val AS t1_val 
FROM t1 
WHERE t1.val = %(val_1)s
INFO:aiopg:{'val_1': 'abc'}
1 abc
2 abc
3 abc

[Pliner]

Hello, @Pliner ! Sure!

pip install aiopg==1.3.2b2
Collecting aiopg==1.3.2b2
  Downloading https://files.pythonhosted.org/packages/4a/f1/1b9a64e90dae1c3ede539d9e665c26f0b721f0fb1ad84ef85fc5b9342126/aiopg-1.3.2b2-py3-none-any.whl
Requirement already satisfied: async-timeout<4.0,>=3.0 in ./lib/python3.8/site-packages (from aiopg==1.3.2b2) (3.0.1)
Requirement already satisfied: psycopg2-binary>=2.8.4 in ./lib/python3.8/site-packages (from aiopg==1.3.2b2) (2.8.5)
Installing collected packages: aiopg
  Found existing installation: aiopg 1.3.1
    Uninstalling aiopg-1.3.1:
      Successfully uninstalled aiopg-1.3.1
Successfully installed aiopg-1.3.2b2

with the code below

import logging
import asyncio
from aiopg.sa import create_engine
import sqlalchemy as sa

logging.basicConfig(level=logging.INFO)

metadata = sa.MetaData()

tbl = sa.Table('t1', metadata,
               sa.Column('id', sa.Integer, primary_key=True),
               sa.Column('val', sa.String(255)))

async def go():
    async with create_engine(user='aiopg',
                             database='aiopg',
                             host='127.0.0.1',
                             password='aiopg', echo=True) as engine:
        async with engine.acquire() as conn:
            await conn.execute(tbl.insert().values(val='abc'))

            async for row in conn.execute(tbl.select(use_labels=True).where(tbl.c.val=='abc')):
                print(row.t1_id, row.t1_val)

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

running with SQLAlchemy-1.4.25 and aiopg-1.3.2b2 there is no problem

INFO:aiopg:SELECT t.oid, typarray FROM pg_type t JOIN pg_namespace ns ON typnamespace = ns.oid WHERE typname = 'hstore';
INFO:aiopg:None
INFO:aiopg:INSERT INTO t1 (val) VALUES (%(val)s) RETURNING t1.id
INFO:aiopg:{'val': 'abc'}
INFO:aiopg:SELECT t1.id AS t1_id, t1.val AS t1_val 
FROM t1 
WHERE t1.val = %(val_1)s
INFO:aiopg:{'val_1': 'abc'}
1 abc
2 abc
3 abc

Thanks. Then, let's release it :)

[Pliner]

https://pypi.org/project/aiopg/1.3.2/