如何在SQLAlchemy中嵌入未支持的SQL原生语法（如ClickHouse WITH FILL）

方法1：结合text和SQLAlchemy子查询嵌入原生SQL #

利用SQLAlchemy的text()构造带WITH FILL的查询，通过:subquery占位符引用之前生成的JOIN子查询，既保留SQLAlchemy的ORM能力，又嵌入原生特性，无需手动拼接字符串。

代码示例：

from sqlalchemy import create_engine, select, table, column, Integer, text
from sqlalchemy.orm import Session

table_x = table(
    'table_x',
    column('id', Integer),
    column('x', Integer)
)
table_y = table(
    'table_y',
    column('id', Integer),
    column('y', Integer)
)

# 初始化引擎和会话
engine = create_engine(f'clickhouse://default:@{clickhouse_host}:{8123}/{database}')
session = Session(engine)

# 1. 用SQLAlchemy构建JOIN后的子查询
joined_table = select(
    table_x.c.id,
    (table_x.c.x + table_y.c.y).label('z')
).select_from(
    table_x.outerjoin(table_y, table_x.c.id == table_y.c.id, full=True)
).subquery()

# 2. 用text构造带WITH FILL的查询，引用joined_table
filled_query = text("""
    SELECT id, z
    FROM :subquery
    ORDER BY z
    WITH FILL STEP 10
""").columns(id=Integer, z=Integer).bindparams(subquery=joined_table)

# 将filled_query转为子查询，供后续SQLAlchemy操作使用
filled_table = select(filled_query.c.id, filled_query.c.z).subquery()

# 3. 用SQLAlchemy构建最终WHERE过滤
final_query = select(filled_table.c.id, filled_table.c.z).where(filled_table.c.z < 200)

# 打印生成的SQL
print(str(final_query.compile(engine, compile_kwargs={"literal_binds": True})))

方法2：自定义ClickHouse编译扩展（更优雅） #

如果需要频繁使用WITH FILL，可以扩展SQLAlchemy的编译逻辑，给Select对象添加自定义方法，让代码更贴合SQLAlchemy的使用习惯。

步骤1：定义自定义编译扩展 #

from sqlalchemy.ext.compiler import compiles
from sqlalchemy.sql.selectable import Select

class WithFillStep:
    def __init__(self, step):
        self.step = step

# 给Select添加with_fill_step方法
def with_fill_step(self, step):
    self._with_fill_step = WithFillStep(step)
    return self

Select.with_fill_step = with_fill_step

# 编译扩展，在SELECT语句末尾添加WITH FILL STEP
@compiles(Select, 'clickhouse')
def compile_select_with_fill(element, compiler, **kwargs):
    sql = compiler.visit_select(element, **kwargs)
    if hasattr(element, '_with_fill_step'):
        fill_step = element._with_fill_step
        sql += f" WITH FILL STEP {fill_step.step}"
    return sql

步骤2：使用自定义方法构建查询 #

# 1. 构建JOIN子查询（和之前一致）
joined_table = select(
    table_x.c.id,
    (table_x.c.x + table_y.c.y).label('z')
).select_from(
    table_x.outerjoin(table_y, table_x.c.id == table_y.c.id, full=True)
).subquery()

# 2. 使用自定义with_fill_step方法添加原生特性
filled_table = select(joined_table.c.id, joined_table.c.z)\
    .order_by(joined_table.c.z)\
    .with_fill_step(10)\
    .subquery()

# 3. 构建最终查询
final_query = select(filled_table.c.id, filled_table.c.z).where(filled_table.c.z < 200)

# 打印SQL
print(str(final_query.compile(engine, compile_kwargs={"literal_binds": True})))

为什么with_hint没用？ #

with_hint是用来给数据库优化器提供表级别的执行提示（比如MySQL的USE INDEX），并非用于添加WITH FILL这种语句级别的原生特性，所以无法满足需求。

内容的提问来源于stack exchange，提问作者Peter