如何用Jinja2和SQLAlchemy在Python中动态生成带多参数的原生SQL

在Python中使用Jinja2和SQLAlchemy对接PostgreSQL数据库，动态生成查询语句，需要获取数据库实际执行的原生SQL字符串。要求通用解决方案，不使用正则替换:age这类参数，也不解析终端日志，希望直接在Python会话中得到可写入文件的字符串。

当前错误 #

raise exc.CompileError(
sqlalchemy.exc.CompileError: No literal value renderer is available for literal value "['New York', 'Boston']" with datatype NULL

期望输出示例 #

WHERE 1 = 1
AND age = 30
AND array_column @> ARRAY['New York', 'Boston']

Jinja模板（query.sql.j2） #

-- query.sql.j2
SELECT * FROM tbl_with_array
WHERE 1 = 1
{% if age %}
AND age = :age
{% endif %}
{% if template_value %}
AND array_column @> :template_value::text[]
{% endif %}

原始Python代码 #

from sqlalchemy.sql import text
import sqlalchemy
from sqlalchemy import create_engine, Table, Column, Integer, String, ARRAY, MetaData
from sqlalchemy.orm import sessionmaker
import jinja2

# (replace with your actual credentials and database details)
engine = create_engine('postgresql+psycopg://user:password@localhost/db', echo=True)
metadata = MetaData()

# Setup example data
tbl_with_array = Table('tbl_with_array', metadata,
                      Column('id', Integer, primary_key=True),
                      Column('age', Integer),
                      Column('array_column', ARRAY(String))
                     )

metadata.create_all(engine)
with engine.connect() as conn:
    conn.execute(tbl_with_array.insert(), [
        {'age': 30, 'array_column': ['New York', 'Boston']},
        {'age': 25, 'array_column': ['London', 'Manchester']},
        {'age': 35, 'array_column': ['San Francisco', 'Los Angeles']}
    ])
    conn.commit()
Session = sessionmaker(bind=engine)
session = Session()

# Jinja2 template
template_env = jinja2.Environment(loader=jinja2.FileSystemLoader(searchpath="."))
template = template_env.get_template('query.sql.j2')
params = {
    'age': 30,
    'template_value': ['New York', 'Boston']
}

# Trying to render the jinja template with the given parameters here.
rendered_query = template.render(age=params.get('age'), template_value=params.get('template_value'))
stmt = text(rendered_query).bindparams(age=params['age'], template_value=params['template_value'])
compiled_stmt = stmt.compile(compile_kwargs={"literal_binds": True})
raw_sql_query = str(compiled_stmt)

result = session.execute(stmt)
for row in result:
    print(row)
print(raw_sql_query)

依赖版本 #

jinja2                   3.1.2        
psycopg                  3.1.13       
sqlalchemy               2.0.23       

错误根源是SQLAlchemy无法自动识别数组参数的类型，编译时将其标记为NULL类型，导致无法生成正确的字面量渲染规则。解决核心是在绑定参数时明确指定数组对应的SQLAlchemy类型，让编译器知道如何将Python列表转换为PostgreSQL的数组语法。

修改后的核心代码片段：

# 确保导入所需类型
from sqlalchemy import ARRAY, String

# ... 其余代码保持不变 ...

# 绑定参数时显式指定类型
stmt = text(rendered_query).bindparams(
    age=params['age'],
    template_value=sqlalchemy.bindparam('template_value', value=params['template_value'], type_=ARRAY(String))
)

# 编译时绑定引擎，自动适配PostgreSQL方言
compiled_stmt = stmt.compile(
    bind=engine,
    compile_kwargs={"literal_binds": True}
)
raw_sql_query = str(compiled_stmt)

关键说明 #

1. 显式参数类型：通过bindparam()声明template_value的类型为ARRAY(String)，让SQLAlchemy明确该参数对应PostgreSQL的text[]类型，从而生成符合要求的ARRAY['New York', 'Boston']语法。
2. 方言适配：编译时传入bind=engine，SQLAlchemy会自动调用PostgreSQL的方言规则渲染字面量，避免跨数据库语法差异。
3. 保留动态逻辑：无需修改Jinja模板的条件判断逻辑，依然可以动态生成SQL片段，兼顾灵活性和正确性。

修改后运行代码，即可得到符合期望的原生SQL字符串，同时正常执行数据库查询。

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