sqlalchemy session.execute with positional arguments error

Question

I am trying to use positional arguments to be used in a lengthy insert query. But it doesnt seem to allow using positional arguments. I am using mysql. Below is the error

sqlalchemy.exc.ArgumentError: List argument must consist only of tuples or dictionaries

sample query with placeholders

        query = "insert into test values(?,?,?)" 
        values = (1,2,3)

tried

        connection = session.connection()
        result = connection.execute(text(query), *values)

        result = session.execute(text(query), *values)

        result = connection.execute(text(query), values)

        result = session.execute(text(query), values)

        changed placeholder.
        query = "insert into test values(%s,%s,%s)"

As the actual query is much bigger , I prefer not to use dict/named arguments

snakecharmerb · Accepted Answer · 2025-01-20 17:55:09Z

2

SQLALchemy 2.0+ does not support this functionality for the execute method of either sessions or connections. However it is possible to use a connection's exec_driver_sql method in the same way as you would execute a query with a DB-API cursor directly.

import sqlalchemy as sa
engine = sa.create_engine('mariadb+pymysql:///test', echo=True)
tbl = sa.Table('users', sa.MetaData(), autoload_with=engine)

# Named arguments (for example %(name)s ) work too, if they are 
# passed in a dictionary.
q = """SELECT id, name FROM users WHERE id = %s or name = %s"""

with engine.connect() as conn:
    rows = conn.exec_driver_sql(q, (1, 'Bob'))
    for row in rows:
        print(row)

If you are using a session, you can use Session.connection() to obtain a reference to the session's connection. This snippet shows how, and logs the connection object to demonstrate that the ORM and exec_driver_sql use the same connection, to avoid transaction isolation issues.

import sqlalchemy as sa
from sqlalchemy import orm

engine = sa.create_engine('mariadb+pymysql:///test', echo=False)

# Set up a simple model mapped to an existing table.
metadata = sa.MetaData()
tbl = sa.Table('users', metadata, autoload_with=engine)

registry = orm.registry(metadata=metadata)

class User:
    pass

registry.map_imperatively(User, tbl)


@sa.event.listens_for(sa.Engine, 'before_cursor_execute', retval=False)
def log_connection_object(conn, cursor, statement, parameters, context, executemany):
    print(f'{conn=}')


q = """SELECT id, name FROM users WHERE id = %s or name = %s"""

with orm.Session(engine) as s:
    users = s.scalars(
        sa.select(User).where(sa.or_(User.id == 1, User.name == 'Bob'))
    ).all()
    rows = s.connection().exec_driver_sql(q, (1, 'Bob'))

The output shows both executions use the same connection:

conn=<sqlalchemy.engine.base.Connection object at 0x7ff475bbe240>
conn=<sqlalchemy.engine.base.Connection object at 0x7ff475bbe240>

edited Jan 20 at 17:55

answered Feb 9, 2024 at 9:17

snakecharmerb

57.1k13 gold badges136 silver badges200 bronze badges

Sign up to request clarification or add additional context in comments.

2 Comments

sjd Over a year ago

Wow !!. Thanks a lot.!! @snakecharmerb. ChatGPT cannot replace SO :D.. Btw do you think if there is any down side of using this function. as obviously they omitted this option in execute method.

snakecharmerb Over a year ago

As long as you use the same connection as the session (I've edited the answer to show how) then there shouldn't be any issues (using a different connection might result in data inconsistencies due to transaction isolation). I suspect Session.execute doesn't support this in order to keep it simple.

Collectives™ on Stack Overflow

sqlalchemy session.execute with positional arguments error

1 Answer 1

2 Comments

Your Answer

Linked

Hot Network Questions

Collectives™ on Stack Overflow

1 Answer 1

2 Comments

Your Answer

Sign up or log in

Post as a guest

Linked

Related