1

我正在使用没有 ORM 的 SQLAlchemy 核心功能，我需要通过将列表作为参数传递的主键从任意表中加载记录。

目前我正在这样做：

records = select([arbitrary_table], list(arbitrary_table.primary_key.columns._all_cols)[0].in_([1, 2, 3]))

arbitrary_table作为参数传递，它可以是任何限制主键是整数而不是复合的表，但主键可以有不同的名称。

我有几个问题：

Q1： 有什么办法可以优化吗？我很确定这list(self.arbitrary_table.primary_key.columns._all_cols)[0]不是获取主键列的最佳方法。

Q2： 如何对 ORM 查询做同样的事情？SA 站点的示例具有my_user = session.query(User).get(5)但它仅将一个 ID 作为参数，并且没有覆盖以获取 ID 列表。

我将非常感谢您的建议。

4

1 回答 1

2

“primary_key”属性是 的一个实例PrimaryKeyConstraint，它可以就地迭代，因此list(table.primary_key)[0]您需要在该列中获取所有内容。

下面的例子说明了两种复合兼容技术（编辑：哦，你说“不是复合”。好吧，无论如何它都是很酷的代码......使用 select_by_single_pk()），一种需要元组支持（postgresql，也许是 mysql），另一种是字符串出 AND/OR，以及单一 PK 方法。然后它显示接受元组的 query.get()：

from sqlalchemy import tuple_, or_, and_

def select_by_composite_pk(table, values):
    "works only in a high-capability database like postgresql"
    return table.select().where(tuple_(*table.primary_key).in_(values))

def select_by_composite_pk_no_tuples(table, values):
    "works in any database"

    return table.select().where(
            or_(
                *[
                    and_(*[col == val for col, val in zip(table.primary_key, val)])
                    for val in values
                ]
            ))

def select_by_single_pk(table, values):
    "works in any database"
    return table.select().where(list(table.primary_key)[0].in_(values))

if __name__ == '__main__':
    from sqlalchemy import create_engine, Table, Column, Integer, MetaData

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

    conn = eng.connect()
    trans = conn.begin()

    m = MetaData()

    # single PK column
    a = Table('a', m, Column('x', Integer, primary_key=True),
                            Column('y', Integer))

    # composite PK column
    b = Table('b', m, Column('x', Integer, primary_key=True),
                    Column('y', Integer, primary_key=True))

    m.create_all(conn)

    conn.execute(a.insert(), [
            {'x': i, 'y': i * 2} for i in xrange(10)
        ])
    conn.execute(b.insert(), [
            {'x': i, 'y': i * 2} for i in xrange(10)
        ])

    print conn.execute(
                select_by_composite_pk(a,
                    [tuple_(3, ), tuple_(5, ), tuple_(9, )])).fetchall()

    print conn.execute(
                select_by_composite_pk(b,
                    [tuple_(3, 6), tuple_(5, 10), tuple_(9, 18)])).fetchall()

    print conn.execute(
                select_by_composite_pk_no_tuples(b,
                    [(3, 6), (5, 10), (9, 18)])).fetchall()

    print conn.execute(
                select_by_single_pk(b, [3, 5, 9])).fetchall()

    # ORM query version
    from sqlalchemy.orm import Session
    from sqlalchemy.ext.declarative import declarative_base

    Base = declarative_base()

    class MyCompositeClass(Base):
        __table__ = b

    # get accepts a tuple
    print Session(conn).query(MyCompositeClass).get((5, 10))
于 2013-03-12T01:58:28.163 回答