SQLAlchemy does not autoconvert type when using `asyncpg` #6365

Ahsoka · 2021-04-26T18:51:47Z

Ahsoka
Apr 26, 2021

Describe the bug

Whenever I use SQLAlchemy with asyncpg and I input the incorrect type SQLAlchemy does not attempt to convert the type to the correct type, unlike when I am using SQLAlchemy with psycopg2. See the examples below for reference. I am not sure if this intended behavior or not.

To Reproduce

Using `psycopg2`

create_table = '''
CREATE TABLE dtype_test (
    index BIGINT,
    "A" FLOAT(53),
    "B" TEXT
)
'''

url = ...
engine = create_engine(url)

with engine.begin() as conn:
    conn.execute(text(create_table))
    # "B" should be an str but here it's boolean
    # SQLAlchemy or psycopg2 automatically converts to an str
    conn.execute(
        text('INSERT INTO dtype_test (index, "A", "B") VALUES(:index, :a, :b)'),
        {'index': 0, 'a': 0.8, 'b': True}
    )
# Runs with no issue

Using `asyncpg`

url = ...

create_table = '''
CREATE TABLE dtype_test (
    index BIGINT,
    "A" FLOAT(53),
    "B" TEXT
)
'''

engine = create_async_engine(url)
async with engine.begin() as conn:
    await conn.execute(text(create_table))
    # Using the wrong type here for "B"
    await conn.execute(text(
        'INSERT INTO dtype_test (index, "A", "B") VALUES(:index, :a, :b)'),
        {'index': 0, 'a': 0.8, 'b': True}
    )
# This errors out, see error below

Error

Traceback (most recent call last):
  File "asyncpg\protocol\prepared_stmt.pyx", line 155, in asyncpg.protocol.protocol.PreparedStatementState._encode_bind_msg
  File "asyncpg\protocol\codecs/base.pyx", line 193, in asyncpg.protocol.protocol.Codec.encode
  File "asyncpg\protocol\codecs/base.pyx", line 104, in asyncpg.protocol.protocol.Codec.encode_scalar
  File "asyncpg\pgproto\./codecs/text.pyx", line 29, in asyncpg.pgproto.pgproto.text_encode
  File "asyncpg\pgproto\./codecs/text.pyx", line 12, in asyncpg.pgproto.pgproto.as_pg_string_and_size
TypeError: expected str, got bool

The above exception was the direct cause of the following exception:

Traceback (most recent call last):
  File "C:\ProgramData\Anaconda3\lib\site-packages\sqlalchemy\dialects\postgresql\asyncpg.py", line 412, in _prepare_and_execute
    self._rows = await prepared_stmt.fetch(*parameters)
  File "C:\ProgramData\Anaconda3\lib\site-packages\asyncpg\prepared_stmt.py", line 168, in fetch
    data = await self.__bind_execute(args, 0, timeout)
  File "C:\ProgramData\Anaconda3\lib\site-packages\asyncpg\prepared_stmt.py", line 233, in __bind_execute
    data, status, _ = await self.__do_execute(
  File "C:\ProgramData\Anaconda3\lib\site-packages\asyncpg\prepared_stmt.py", line 222, in __do_execute
    return await executor(protocol)
  File "asyncpg\protocol\protocol.pyx", line 181, in bind_execute
  File "asyncpg\protocol\prepared_stmt.pyx", line 171, in asyncpg.protocol.protocol.PreparedStatementState._encode_bind_msg
asyncpg.exceptions.DataError: invalid input for query argument $3: True (expected str, got bool)

The above exception was the direct cause of the following exception:

Traceback (most recent call last):
  File "C:\ProgramData\Anaconda3\lib\site-packages\sqlalchemy\engine\base.py", line 1706, in _execute_context
    self.dialect.do_execute(
  File "C:\ProgramData\Anaconda3\lib\site-packages\sqlalchemy\engine\default.py", line 716, in do_execute
    cursor.execute(statement, parameters)
  File "C:\ProgramData\Anaconda3\lib\site-packages\sqlalchemy\dialects\postgresql\asyncpg.py", line 449, in execute
    self._adapt_connection.await_(
  File "C:\ProgramData\Anaconda3\lib\site-packages\sqlalchemy\util\_concurrency_py3k.py", line 61, in await_only
    return current.driver.switch(awaitable)
  File "C:\ProgramData\Anaconda3\lib\site-packages\sqlalchemy\util\_concurrency_py3k.py", line 114, in greenlet_spawn
    value = await result
  File "C:\ProgramData\Anaconda3\lib\site-packages\sqlalchemy\dialects\postgresql\asyncpg.py", line 424, in _prepare_and_execute
    self._handle_exception(error)
  File "C:\ProgramData\Anaconda3\lib\site-packages\sqlalchemy\dialects\postgresql\asyncpg.py", line 358, in _handle_exception
    self._adapt_connection._handle_exception(error)
  File "C:\ProgramData\Anaconda3\lib\site-packages\sqlalchemy\dialects\postgresql\asyncpg.py", line 652, in _handle_exception
    raise translated_error from error
sqlalchemy.dialects.postgresql.asyncpg.Error: <class 'asyncpg.exceptions.DataError'>: invalid input for query argument $3: True (expected str, got bool)

The above exception was the direct cause of the following exception:

Traceback (most recent call last):
  File "error.py", line 42, in <module>
    asyncio.run(main())
  File "C:\ProgramData\Anaconda3\lib\asyncio\runners.py", line 43, in run
    return loop.run_until_complete(main)
  File "C:\ProgramData\Anaconda3\lib\asyncio\base_events.py", line 616, in run_until_complete
    return future.result()
  File "error.py", line 40, in main
    await conn.execute(text(
  File "C:\ProgramData\Anaconda3\lib\site-packages\sqlalchemy\ext\asyncio\engine.py", line 411, in execute
    result = await greenlet_spawn(
  File "C:\ProgramData\Anaconda3\lib\site-packages\sqlalchemy\util\_concurrency_py3k.py", line 119, in greenlet_spawn
    result = context.throw(*sys.exc_info())
  File "C:\ProgramData\Anaconda3\lib\site-packages\sqlalchemy\engine\base.py", line 1521, in _execute_20
    return meth(self, args_10style, kwargs_10style, execution_options)
  File "C:\ProgramData\Anaconda3\lib\site-packages\sqlalchemy\sql\elements.py", line 313, in _execute_on_connection
    return connection._execute_clauseelement(
  File "C:\ProgramData\Anaconda3\lib\site-packages\sqlalchemy\engine\base.py", line 1390, in _execute_clauseelement
    ret = self._execute_context(
  File "C:\ProgramData\Anaconda3\lib\site-packages\sqlalchemy\engine\base.py", line 1749, in _execute_context
    self._handle_dbapi_exception(
  File "C:\ProgramData\Anaconda3\lib\site-packages\sqlalchemy\engine\base.py", line 1930, in _handle_dbapi_exception
    util.raise_(
  File "C:\ProgramData\Anaconda3\lib\site-packages\sqlalchemy\util\compat.py", line 211, in raise_
    raise exception
  File "C:\ProgramData\Anaconda3\lib\site-packages\sqlalchemy\engine\base.py", line 1706, in _execute_context
    self.dialect.do_execute(
  File "C:\ProgramData\Anaconda3\lib\site-packages\sqlalchemy\engine\default.py", line 716, in do_execute
    cursor.execute(statement, parameters)
  File "C:\ProgramData\Anaconda3\lib\site-packages\sqlalchemy\dialects\postgresql\asyncpg.py", line 449, in execute
    self._adapt_connection.await_(
  File "C:\ProgramData\Anaconda3\lib\site-packages\sqlalchemy\util\_concurrency_py3k.py", line 61, in await_only
    return current.driver.switch(awaitable)
  File "C:\ProgramData\Anaconda3\lib\site-packages\sqlalchemy\util\_concurrency_py3k.py", line 114, in greenlet_spawn
    value = await result
  File "C:\ProgramData\Anaconda3\lib\site-packages\sqlalchemy\dialects\postgresql\asyncpg.py", line 424, in _prepare_and_execute
    self._handle_exception(error)
  File "C:\ProgramData\Anaconda3\lib\site-packages\sqlalchemy\dialects\postgresql\asyncpg.py", line 358, in _handle_exception
    self._adapt_connection._handle_exception(error)
  File "C:\ProgramData\Anaconda3\lib\site-packages\sqlalchemy\dialects\postgresql\asyncpg.py", line 652, in _handle_exception
    raise translated_error from error
sqlalchemy.exc.DBAPIError: (sqlalchemy.dialects.postgresql.asyncpg.Error) <class 'asyncpg.exceptions.DataError'>: invalid input for query argument $3: True (expected str, got bool)
[SQL: INSERT INTO dtype_test (index, "A", "B") VALUES(%s, %s, %s)]
[parameters: (0, 0.8, True)]
(Background on this error at: http://sqlalche.me/e/14/dbapi)

Versions.

OS: Windows 10
Python: 3.8.5
SQLAlchemy: 1.4.11
Database: PostgreSQL
DBAPI: asyncpg and psycopg2

Have a nice day!

zzzeek · 2021-04-26T19:13:08Z

zzzeek
Apr 26, 2021
Maintainer

SQLAlchemy doesn't infer types when they are passed to connection.execute(), it only does its own inference when SQL constructs are being constructed, such as if you say something like select(table.c.x == 5).

if asyncpg is complaining about types you would have to supply these upfront. you can send the values directly to text.bindparams() where it will infer the types:

stmt = text(
     'INSERT INTO dtype_test (index, "A", "B") VALUES(:index, :a, :b)'
).bindparams(index=0, a=0.8, b=True)

you could use those values as placeholders, they will be overridden if you supply different values to .execute().

or you can send just the datatypes, but the syntax is more verbose at the moment:

stmt = text(
     'INSERT INTO dtype_test (index, "A", "B") VALUES(:index, :a, :b)'
).bindparams(bindparam("index", type_=Integer), bindparam("a", type_=Float), bindparam("b", type_=Boolean)))

10 replies

CaselIT Apr 27, 2021
Maintainer

I think it's important to note that if you are sending Python boolean "True" into a string and it's magically becoming something, that is a bug in those drivers. For SQLite, we know this is how SQLite rolls. but for psycopg2 it's just wrong. psycopg2 seems to generate the string "true".

Actually that's documented by psycopg2 here: https://www.psycopg.org/docs/usage.html#adaptation-of-python-values-to-sql-types. What I think is happening with True is that psycopg2 adapts it to the value true, that is the bool representation in pg, and that value is inserted as string by the db. So I don't think there is a bug there.

So it seems that Psycopg2 uses the input values to perform the conversion, while asyncpg works differently, by preparing the query, getting the type information from the db and then casting the input values to what the db espects.

CaselIT Apr 27, 2021
Maintainer

Indeed the following sql works as the example:

begin; 
create table foo (value text);
insert into foo values (true);
select * from foo;
rollback;

The select returns

value|
-----|
true |

Ahsoka Apr 27, 2021
Author

Ok new question. My example above is actually not representative of how Pandas creates and inserts data from a Pandas DataFrame into an SQL table, at least with regards to the to_sql method. Please see the following example for how this operation in conducted (this is a simplified version of the actual operation):

url = ...

async_engine = create_async_engine(url)

async with async_engine.begin() as conn:
    values = [{'index': 0, 'A': 0.8, 'B': True}, {'index': 1, 'A': 0.9, 'B': None}]

    meta = MetaData()
    table = Table(
        "dtype_test",
        meta,
        Column('index', BigInteger),
        Column('A', Float(53)),
        Column('B', Text) # This column would normally be a Boolean column but here the user overrides it
    )
    await conn.run_sync(table.create)
    await conn.execute(table.insert(), parameters=values)

Once again we get the same error as before:

Traceback (most recent call last):
  File "asyncpg\protocol\prepared_stmt.pyx", line 155, in asyncpg.protocol.protocol.PreparedStatementState._encode_bind_msg
  File "asyncpg\protocol\codecs/base.pyx", line 193, in asyncpg.protocol.protocol.Codec.encode
  File "asyncpg\protocol\codecs/base.pyx", line 104, in asyncpg.protocol.protocol.Codec.encode_scalar
  File "asyncpg\pgproto\./codecs/text.pyx", line 29, in asyncpg.pgproto.pgproto.text_encode
  File "asyncpg\pgproto\./codecs/text.pyx", line 12, in asyncpg.pgproto.pgproto.as_pg_string_and_size
TypeError: expected str, got bool

The above exception was the direct cause of the following exception:

Traceback (most recent call last):
  File "C:\ProgramData\Anaconda3\lib\site-packages\sqlalchemy\dialects\postgresql\asyncpg.py", line 442, in _executemany
    return await self._connection.executemany(
  File "C:\ProgramData\Anaconda3\lib\site-packages\asyncpg\connection.py", line 337, in executemany
    return await self._executemany(command, args, timeout)
  File "C:\ProgramData\Anaconda3\lib\site-packages\asyncpg\connection.py", line 1663, in _executemany
    result, _ = await self._do_execute(query, executor, timeout)
  File "C:\ProgramData\Anaconda3\lib\site-packages\asyncpg\connection.py", line 1697, in _do_execute
    result = await executor(stmt, None)
  File "asyncpg\protocol\protocol.pyx", line 252, in bind_execute_many
  File "asyncpg\protocol\coreproto.pyx", line 945, in asyncpg.protocol.protocol.CoreProtocol._bind_execute_many_more
  File "asyncpg\protocol\protocol.pyx", line 218, in genexpr
  File "asyncpg\protocol\prepared_stmt.pyx", line 171, in asyncpg.protocol.protocol.PreparedStatementState._encode_bind_msg
asyncpg.exceptions.DataError: invalid input for query argument $3: True (expected str, got bool)

The above exception was the direct cause of the following exception:

Traceback (most recent call last):
  File "C:\ProgramData\Anaconda3\lib\site-packages\sqlalchemy\engine\base.py", line 1686, in _execute_context
    self.dialect.do_executemany(
  File "C:\ProgramData\Anaconda3\lib\site-packages\sqlalchemy\engine\default.py", line 713, in do_executemany
    cursor.executemany(statement, parameters)
  File "C:\ProgramData\Anaconda3\lib\site-packages\sqlalchemy\dialects\postgresql\asyncpg.py", line 454, in executemany
    return self._adapt_connection.await_(
  File "C:\ProgramData\Anaconda3\lib\site-packages\sqlalchemy\util\_concurrency_py3k.py", line 61, in await_only
    return current.driver.switch(awaitable)
  File "C:\ProgramData\Anaconda3\lib\site-packages\sqlalchemy\util\_concurrency_py3k.py", line 114, in greenlet_spawn
    value = await result
  File "C:\ProgramData\Anaconda3\lib\site-packages\sqlalchemy\dialects\postgresql\asyncpg.py", line 446, in _executemany
    self._handle_exception(error)
  File "C:\ProgramData\Anaconda3\lib\site-packages\sqlalchemy\dialects\postgresql\asyncpg.py", line 358, in _handle_exception
    self._adapt_connection._handle_exception(error)
  File "C:\ProgramData\Anaconda3\lib\site-packages\sqlalchemy\dialects\postgresql\asyncpg.py", line 652, in _handle_exception
    raise translated_error from error
sqlalchemy.dialects.postgresql.asyncpg.Error: <class 'asyncpg.exceptions.DataError'>: invalid input for query argument $3: True (expected str, got bool)

The above exception was the direct cause of the following exception:

Traceback (most recent call last):
  File "error.py", line 73, in <module>
    asyncio.run(main())
  File "C:\ProgramData\Anaconda3\lib\asyncio\runners.py", line 43, in run
    return loop.run_until_complete(main)
  File "C:\ProgramData\Anaconda3\lib\asyncio\base_events.py", line 616, in run_until_complete
    return future.result()
  File "error.py", line 69, in main
    await conn.execute(table.insert(), parameters=values)
  File "C:\ProgramData\Anaconda3\lib\site-packages\sqlalchemy\ext\asyncio\engine.py", line 411, in execute
    result = await greenlet_spawn(
  File "C:\ProgramData\Anaconda3\lib\site-packages\sqlalchemy\util\_concurrency_py3k.py", line 119, in greenlet_spawn
    result = context.throw(*sys.exc_info())
  File "C:\ProgramData\Anaconda3\lib\site-packages\sqlalchemy\engine\base.py", line 1521, in _execute_20
    return meth(self, args_10style, kwargs_10style, execution_options)
  File "C:\ProgramData\Anaconda3\lib\site-packages\sqlalchemy\sql\elements.py", line 313, in _execute_on_connection
    return connection._execute_clauseelement(
  File "C:\ProgramData\Anaconda3\lib\site-packages\sqlalchemy\engine\base.py", line 1390, in _execute_clauseelement
    ret = self._execute_context(
  File "C:\ProgramData\Anaconda3\lib\site-packages\sqlalchemy\engine\base.py", line 1749, in _execute_context
    self._handle_dbapi_exception(
  File "C:\ProgramData\Anaconda3\lib\site-packages\sqlalchemy\engine\base.py", line 1930, in _handle_dbapi_exception
    util.raise_(
  File "C:\ProgramData\Anaconda3\lib\site-packages\sqlalchemy\util\compat.py", line 211, in raise_
    raise exception
  File "C:\ProgramData\Anaconda3\lib\site-packages\sqlalchemy\engine\base.py", line 1686, in _execute_context
    self.dialect.do_executemany(
  File "C:\ProgramData\Anaconda3\lib\site-packages\sqlalchemy\engine\default.py", line 713, in do_executemany
    cursor.executemany(statement, parameters)
  File "C:\ProgramData\Anaconda3\lib\site-packages\sqlalchemy\dialects\postgresql\asyncpg.py", line 454, in executemany
    return self._adapt_connection.await_(
  File "C:\ProgramData\Anaconda3\lib\site-packages\sqlalchemy\util\_concurrency_py3k.py", line 61, in await_only
    return current.driver.switch(awaitable)
  File "C:\ProgramData\Anaconda3\lib\site-packages\sqlalchemy\util\_concurrency_py3k.py", line 114, in greenlet_spawn
    value = await result
  File "C:\ProgramData\Anaconda3\lib\site-packages\sqlalchemy\dialects\postgresql\asyncpg.py", line 446, in _executemany
    self._handle_exception(error)
  File "C:\ProgramData\Anaconda3\lib\site-packages\sqlalchemy\dialects\postgresql\asyncpg.py", line 358, in _handle_exception
    self._adapt_connection._handle_exception(error)
  File "C:\ProgramData\Anaconda3\lib\site-packages\sqlalchemy\dialects\postgresql\asyncpg.py", line 652, in _handle_exception
    raise translated_error from error
sqlalchemy.exc.DBAPIError: (sqlalchemy.dialects.postgresql.asyncpg.Error) <class 'asyncpg.exceptions.DataError'>: invalid input for query argument $3: True (expected str, got bool)
[SQL: INSERT INTO dtype_test (index, "A", "B") VALUES (%s, %s, %s)]
[parameters: ((0, 0.8, True), (1, 0.9, None))]
(Background on this error at: http://sqlalche.me/e/14/dbapi)

Is there anyway to do a bindparams operation with the table.insert() clause, because actually I do have access to the SQLAlchemy types of each column if there are not they expected type — for example if a certain column dtype in Pandas is a bool, the user can override it with by passing the text dtype, and when they override the column type I have access to this overridden type. I believe this is what the unit test is testing.

CaselIT Apr 27, 2021
Maintainer

I think you could use cast if you have access to the table type

Ahsoka Apr 27, 2021
Author

I think you could use cast if you have access to the table type

Do you think you could provide an example?

This is how the table dtype is provided:

async def to_sql(..., dtype=None):
    ...

import sqlalchemy

engine = create_async_engine(...)

await some_dataframe.to_sql('TABLE GOES HERE', engine, dtype={'some_column': sqlalchemy.Text})

CaselIT · 2021-04-26T20:16:37Z

CaselIT
Apr 26, 2021
Maintainer

I think the difference in behaviour is actually on the dbapi level.
In both cases SQLAlchemy passes the value as is to the dbapi, but psycopg2 must be doing some conversion itself.

You can verify this by adding echo=True to the create_engine call.
For psycopg2 it will log something like:

2021-04-26 22:13:54,182 INFO sqlalchemy.engine.Engine INSERT INTO dtype_test (index, "A", "B") VALUES(%(index)s, %(a)s, %(b)s)
2021-04-26 22:13:54,182 INFO sqlalchemy.engine.Engine [generated in 0.00037s] {'index': 0, 'a': 0.8, 'b': True}
2021-04-26 22:13:54,184 INFO sqlalchemy.engine.Engine COMMIT

As you can see True is passed as boolean, not as a string.

0 replies

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

SQLAlchemy does not autoconvert type when using `asyncpg` #6365

{{title}}

Replies: 2 comments 10 replies

{{title}}

{{title}}

{{title}}

{{editor}}'s edit

{{editor}}'s edit

{{title}}

{{editor}}'s edit

{{editor}}'s edit

{{title}}

{{title}}

{{title}}

{{editor}}'s edit

{{editor}}'s edit

Select a reply

SQLAlchemy does not autoconvert type when using asyncpg #6365

Ahsoka Apr 26, 2021

Describe the bug

To Reproduce

Using psycopg2

Using asyncpg

Replies: 2 comments · 10 replies

zzzeek Apr 26, 2021 Maintainer

CaselIT Apr 27, 2021 Maintainer

CaselIT Apr 27, 2021 Maintainer

Ahsoka Apr 27, 2021 Author

CaselIT Apr 27, 2021 Maintainer

Ahsoka Apr 27, 2021 Author

CaselIT Apr 26, 2021 Maintainer

SQLAlchemy does not autoconvert type when using `asyncpg` #6365

Ahsoka
Apr 26, 2021

Using `psycopg2`

Using `asyncpg`

Replies: 2 comments 10 replies

zzzeek
Apr 26, 2021
Maintainer

CaselIT Apr 27, 2021
Maintainer

CaselIT Apr 27, 2021
Maintainer

Ahsoka Apr 27, 2021
Author

CaselIT Apr 27, 2021
Maintainer

Ahsoka Apr 27, 2021
Author

CaselIT
Apr 26, 2021
Maintainer