使用 sqlalchemy WHERE 子句和 IN 运算符时维护顺序

Question

考虑以下数据库table：

ID   ticker   description
1    GDBR30   30YR
2    GDBR10   10YR
3    GDBR5    5YR
4    GDBR2    2YR

可以用这段代码复制：

from sqlalchemy import (
    Column,
    Integer,
    MetaData,
    String,
    Table,
    create_engine,
    insert,
    select,
)

engine = create_engine("sqlite+pysqlite:///:memory:", echo=True, future=True)

metadata = MetaData()

# Creating the table
tickers = Table(
    "tickers",
    metadata,
    Column("id", Integer, primary_key=True, autoincrement=True),
    Column("ticker", String, nullable=False),
    Column("description", String(), nullable=False),
)

metadata.create_all(engine)

# Populating the table
with engine.connect() as conn:
    result = conn.execute(
        insert(tickers),
        [
            {"ticker": "GDBR30", "description": "30YR"},
            {"ticker": "GDBR10", "description": "10YR"},
            {"ticker": "GDBR5", "description": "5YR"},
            {"ticker": "GDBR2", "description": "2YR"},
        ],
    )
    conn.commit()

我需要过滤 tickers 以获得某些值：

search_list = ["GDBR10", "GDBR5", "GDBR30"]

records = conn.execute(
    select(tickers.c.description).where((tickers.c.ticker).in_(search_list))
)

print(records.fetchall())

# Result
# [('30YR',), ('10YR',), ('5YR',)]

但是，我需要按照 search_list 的排序方式生成的元组列表。也就是说，我需要以下结果：

print(records.fetchall())

# Expected result
# [('10YR',), ('5YR',), ('30YR',)]

使用 SQLite，您可以创建一个包含两列（id 和 ticker）的 cte。应用以下代码将产生预期的结果（参见）。不幸的是，我无法将 SQLite 解决方案转移到 sqlalchemy。

WITH cte(id, ticker) AS (VALUES (1, 'GDBR10'), (2, 'GDBR5'), (3, 'GDBR30'))
SELECT t.* 
FROM tbl t INNER JOIN cte c
ON c.ticker = t.ticker
ORDER BY c.id

假设，我有如下 search_list_tuple，我应该如何编写 sqlalchemy 查询的代码？

search_list_tuple = [(1, 'GDBR10'), (2, 'GDBR5'), (3, 'GDBR30')]

Answer 1

下面的工作实际上等同于 sqlite 上的 VALUES (...)，尽管有点冗长：

# construct the CTE
sub_queries = [
    select(literal(i).label("id"), literal(v).label("ticker"))
    for i, v in enumerate(search_list)
]
cte = union_all(*sub_queries).cte("cte")

# desired query
records = conn.execute(
    select(tickers.c.description)
    .join(cte, cte.c.ticker == tickers.c.ticker)
    .order_by(cte.c.id)
)
print(records.fetchall())
# [('10YR',), ('5YR',), ('30YR',)]

下面是使用 values() 结构，但不幸的是，结果查询在 SQLite 上失败，但它在 postgresql 上运行完美：

cte = select(
    values(
        column("id", Integer), column("ticker", String), name="subq"
    ).data(list(zip(range(len(search_list)), search_list)))
).cte("cte")

qq = (
    select(tickers.c.description)
    .join(cte, cte.c.ticker == tickers.c.ticker)
    .order_by(cte.c.id)
)
records = conn.execute(qq)
print(records.fetchall())

使用 sqlalchemy WHERE 子句和 IN 运算符时维护顺序

Maintain order when using sqlalchemy WHERE-clause and IN operator

python

sqlalchemy