如何使用SQLAlchemy方言类型的literal_processor方法？

方法核心作用 #

literal_processor 是SQLAlchemy类型系统的底层方法，用于将Python字面量值直接渲染为目标数据库兼容的SQL片段。它不会被开发者直接调用，而是由SQLAlchemy的查询编译引擎在特定场景下自动触发。

ClickHouse Array类型示例分析 #

以clickhouse_sqlalchemy.types.Array的实现为例：

def literal_processor(self, dialect):
    item_processor = self.item_type_impl.literal_processor(dialect)

    def process(value):
        processed_value = []
        for x in value:
            if item_processor:
                x = item_processor(x)
            processed_value.append(x)
        return '[' + ', '.join(processed_value) + ']'
    return process

该方法返回的process函数，会将Python列表转换为ClickHouse原生数组格式（如[1, 2, 3]）。

触发逻辑生成原生数组的实用方式 #

无需手动调用literal_processor，以下两种场景会自动触发该逻辑：

1. 使用literal函数传入Python列表 #

通过sqlalchemy.sql.expression.literal传入Python列表并指定Array类型，SQLAlchemy会自动调用处理器生成原生数组：

from sqlalchemy import literal, select
from clickhouse_sqlalchemy.types import Array, String

# 生成ClickHouse原生数组 [ 'a', 'b', 'c' ]
array_literal = literal(['a', 'b', 'c'], type_=Array(String))
# 用于查询
query = select(array_literal)

2. 绑定参数时指定类型 #

在带参数的查询中，给参数指定Array类型，编译时会自动触发处理器：

from sqlalchemy import text
from clickhouse_sqlalchemy.types import Array, Int64

query = text("SELECT * FROM my_table WHERE id IN UNNEST(:my_array)")
# 传入参数并指定类型
result = session.execute(
    query,
    {"my_array": [1,2,3]},
    execution_options={"type_map": {"my_array": Array(Int64)}}
)

直接编写原生SQL片段（替代方案） #

如果需要更直接的方式，可通过text函数直接写入ClickHouse数组语法：

from sqlalchemy import text, select

array_expr = text("[1, 2, 3]")
query = select(array_expr)

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