在SQLAlchemy中处理大小写敏感列的MERGE更新问题

问题背景 #

我正在为SQL代码编写测试，定义了如下ORM表：

class CompanyAddressORM(Base):
    __tablename__ = "company_address"

    pk = Column("pk", Text, primary_key=True)
    project_id = Column("PROJECT_ID", Integer)
    company_id = Column("COMPANY_ID", Integer)


class CompanyDocumentsORM(Base):
    __tablename__ = "company_documents"

    pk = Column("pk", Text, primary_key=True)
    project_id = Column("PROJECT_ID", Integer)
    company_id = Column("COMPANY_ID", Integer)
    page_id = Column("PAGE_ID", Integer)

底层代码使用Snowflake，无法修改原有代码，需通过testing.postgresql包模拟Snowflake连接执行查询，遇到了大小写敏感列的问题：

• 直接查询大写列名报错：

session.execute("select PROJECT_ID from company_address")
>>> *** sqlalchemy.exc.ProgrammingError: (psycopg2.errors.UndefinedColumn) column "project_id" does not exist

• 给列名加双引号后查询正常：

session.execute('select "PROJECT_ID" from company_address').all()
>>> [(1,), (2,), (3,), (4,), (5,)]

后续执行MERGE更新时问题更突出：原有MERGE在Snowflake中正常运行，但在SQLAlchemy中因大小写不敏感报错。尝试给所有列加双引号：

source_column_names = ", ".join(f'''"source.{col}"''' for col in columns)
target_column_updates = ", ".join(f'''"target.{col}" = "source.{col}"''' for col in columns)

# Generate the join condition for the merge query
join_condition = " AND ".join(
    f'''"target.{col}" = "source.{col}"''' for col in join_columns
)

生成的查询语句如下：

MERGE INTO company_address as target
    USING company_address_tmp as source
    ON "target.project_id" = "source.project_id"
    WHEN MATCHED THEN UPDATE SET
        .....

但执行该查询时出现编译错误；若去掉双引号，又会出现列不存在的错误。

解决方法 #

1. 修正双引号的使用方式 #

你当前的错误是把表别名+列名整体用双引号包裹了，PostgreSQL会把"target.project_id"当成一个完整的列名，而不是target表下的project_id列。正确的做法是只给列名加双引号，表别名不需要：

# 正确的列名拼接方式
source_column_names = ", ".join(f'source."{col}"' for col in columns)
target_column_updates = ", ".join(f'target."{col}" = source."{col}"' for col in columns)

join_condition = " AND ".join(
    f'target."{col}" = source."{col}"' for col in join_columns
)

生成的MERGE语句会变成：

MERGE INTO company_address as target
    USING company_address_tmp as source
    ON target."PROJECT_ID" = source."PROJECT_ID"
    WHEN MATCHED THEN UPDATE SET
        target."COMPANY_ID" = source."COMPANY_ID"
        .....

2. 让SQLAlchemy自动处理大小写敏感列 #

如果不想手动拼接双引号，可以通过SQLAlchemy的配置让它自动为列名添加双引号：

在创建引擎时，开启quote_all_identifiers参数：

from sqlalchemy import create_engine

# 针对testing.postgresql的引擎配置
engine = create_engine(
    postgresql_url,
    connect_args={"options": "-c quote_all_identifiers=on"}
)

开启后，SQLAlchemy会自动给所有标识符（表名、列名）添加双引号，避免大小写转换问题。

3. 模拟Snowflake的大小写行为 #

Snowflake默认保留标识符的大小写（用双引号包裹时），而PostgreSQL默认会把未加双引号的标识符转为小写。可以在测试环境初始化PostgreSQL时，修改数据库参数让它和Snowflake行为对齐：

from testing.postgresql import Postgresql

with Postgresql(
    initdb_args=["--lc-collate=C", "--lc-ctype=C"],
    postgres_args=["-c", "sql_identifier_case=upper"]
) as pg:
    # 在这里创建连接、执行测试
    pass

这样PostgreSQL会把未加双引号的标识符视为大写，和Snowflake的行为一致，无需手动添加双引号。

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