带有IN子句的SQLAlchemy原始SQL参数替换

Question

我有一个SQL语句,SELECT foo FROM bar WHERE id IN %s.我有一个整数列表,例如[1, 2, 3],我希望它变成一个看起来像的SQL语句SELECT foo FROM bar WHERE id IN (1, 2, 3).

我使用SQLAlchemy Core作为其连接池,并使一些带有多个VALUES子句的插入更容易编写和维护.我更喜欢在原始SQL中编写大部分查询.

要在Pyscopg2中这样做我做cursor.execute('SELECT .. WHERE IN %s', (tuple(my_list),)).但是,我无法在SQLAlchemy中完成这项工作.

engine.execute('SELECT ... WHERE IN %s', tuple(my_list))引发异常:TypeError:并非在字符串格式化期间转换所有参数.如果我只传递列表而不包含在元组中,则会引发同样的异常.

如果我使用命名参数,就像engine.execute('SELECT ... WHERE id IN :ids', ids=my_list)我得到一个ProgrammingError异常,因为SQLAlchemy创建了错误的SQL :( SELECT * FROM foo WHERE id IN :ids它不会替换我的变量的:ids值).如果我传递一个元组,则会引发同样的异常.

如何WHERE IN()在SQLAlchemy中使用原始SQL 的子句？

Answer 1

这是一种仅由某些DBAPI支持的异常格式,因为它将一个项目元组呈现为单独的SQL表达式,包括它在参数之间呈现逗号等,因此语句就像execute("select * from table where value in %s", (somelist, ))在数据库级别扩展到select * from table where value in (1, 2, 3).

SQLAlchemy不期望这种格式 - 它已经对传入参数进行了一些检查,因为它涉及将参数路由到DBAPI execute()或executemany()方法,并且还接受几种不同的样式,并且这种转换的结果是这里的元组变得扁平了.你可以通过添加一个元组来隐藏你的元组经过这个解析:

from sqlalchemy import create_engine

engine = create_engine("postgresql://scott:tiger@localhost/test", echo=True)

with engine.connect() as conn:
    trans = conn.begin()


    conn.execute("create table test (data integer)")
    conn.execute(
            "insert into test (data) values (%s)",
            [(1, ), (2, ), (3, ), (4, ), (5, )]
        )

    result = conn.execute(
                "select * from test where data in %s",
                (
                    ((1, 2, 3),),
                )
            )

    print result.fetchall()

以上样式仅适用于某些DBAPI.快速测试确认它适用于psycopg2和MySQLdb,但不适用于sqlite3.它更多地与DBAPI用于将绑定参数发送到数据库的底层系统有关; psycopg2和MySQLdb都进行Python字符串插值和它们自己的转义,但像cx_oracle这样的系统会将参数单独传递给OCI,所以这种情况在这种情况下不起作用.

SQLAlchemy当然在使用SQL表达式构造时提供了in_()运算符,但这不适用于直字符串.

Answer 2

我使用SQLAlchemy 0.9.8,python 2.7,MySQL 5.X和MySQL-Python作为连接器,在这种情况下,需要一个元组.我的代码如下:

id_list = [1, 2, 3, 4, 5] # in most case we have an integer list or set
s = text('SELECT id, content FROM myTable WHERE id IN :id_list')
conn = engine.connect() # get a mysql connection
rs = conn.execute(s, id_list=tuple(id_list)).fetchall()

希望一切都适合你.