控制（或避免？）SQLAlchemy 中的 CTE 别名

Question

我一直在尝试在 SQLAlchemy 1.4 中实现一个简单的“数到三”递归 CTE（常见 table 表达式）。本机 SQL 版本是

WITH RECURSIVE my_cte (i)
AS (
    SELECT 1
    UNION ALL
    SELECT i + 1 FROM my_cte WHERE i < 3
)
SELECT i FROM my_cte ORDER BY i;

哪个returns

 i
--
 1
 2
 3

我最初尝试根据其中一个示例 here 翻译成 SQLAlchemy（核心）是

my_cte = (
    select(literal_column("1").label("i"))
    .cte(name="my_cte", recursive=True)
)
my_cte_alias = my_cte.alias()
my_cte = my_cte.union_all(
    select(
        literal_column("my_cte.i + 1"),
    )
    .where(my_cte_alias.c.i < 3)
)
stmt = select(my_cte.c.i).order_by(my_cte.c.i)
with engine.begin() as conn:
    result = conn.execute(stmt).fetchall()
    print(result)

但它给了我（非常有帮助！）错误

psycopg2.errors.UndefinedTable: invalid reference to FROM-clause entry for table "my_cte"
LINE 2: (SELECT 1 AS i UNION ALL SELECT my_cte.i + 1 
                                        ^
HINT:  Perhaps you meant to reference the table alias "anon_1".

确实，我需要更改我的第二个 literal_column() 以使用 anon_1 而不是 my_cte

my_cte = (
    select(literal_column("1").label("i"))
    .cte(name="my_cte", recursive=True)
)
my_cte_alias = my_cte.alias()
my_cte = my_cte.union_all(
    select(
        literal_column("anon_1.i + 1"),
    )
    .where(my_cte_alias.c.i < 3)
)
stmt = select(my_cte.c.i).order_by(my_cte.c.i)
with engine.begin() as conn:
    result = conn.execute(stmt).fetchall()
    print(result)

这奏效了。但是，最好避免使用“anon_1”别名，或者至少控制所使用的别名，以防更复杂的情况可能有其他可能造成混乱的“anon_x”别名开始工作。

有人可以建议我该怎么做吗？

Answer 1

SQLAlchemy 的 CTE alias 接受一个名称参数，可用于在别处引用别名：

...
# Name the alias.
my_cte_alias = my_cte.alias("my_alias")
my_cte = my_cte.union_all(
    select(
        literal_column("my_alias.i + 1"),  # <- use the name here to refer to the alias.
    )
    .where(my_cte_alias.c.i < 3)
)
...

控制（或避免？）SQLAlchemy 中的 CTE 别名

Control (or avoid?) CTE alias in SQLAlchemy

python

sqlalchemy

alias

common-table-expression