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
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>