pythonsqlalchemy

sqlalchemy session.execute with positional arguments error


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


Solution

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