python - 使用 sqlalchemy WHERE 子句和 IN 运算符时保持顺序

Question

考虑以下数据库表:

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，你可以创建一个 cte有两列( id 和 ticker )。应用以下代码将导致预期结果(请参阅 Maintain order when using SQLite WHERE-clause and IN operator )。不幸的是，我无法将 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

下面的作品实际上相当于VALUES (...)在 sqlite虽然有点冗长:

# 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())