如何自定义适配Alembic+Snowflake的SQLAlchemy建表DDL

SQLAlchemy默认生成小写的schema、表、列标识符，Snowflake对未加引号的标识符默认存储为全大写格式，两边大小写不匹配会直接导致Alembic无法识别已存在的库表，抛出I looked for my_schema.my_table but only found MY_SCHEMA.MY_TABLE类错误，可通过以下方式实现标识符全大写统一，不需要逐个修改模型字段定义：

方案1：DDL编译钩子全局统一转大写（无侵入） #

通过SQLAlchemy的编译扩展拦截所有DDL生成逻辑，自动把schema、表、列名转换为大写，不需要修改现有模型代码：

from sqlalchemy.ext.compiler import compiles
from sqlalchemy.schema import CreateTable, CreateColumn, CreateIndex

@compiles(CreateTable)
def uppercase_create_table(element, compiler, **kw):
    # 转换表名、schema名大写
    table = element.element
    table.name = table.name.upper()
    if table.schema:
        table.schema = table.schema.upper()
    return compiler.visit_create_table(element, **kw)

@compiles(CreateColumn)
def uppercase_create_column(element, compiler, **kw):
    # 转换列名大写
    element.element.name = element.element.name.upper()
    return compiler.visit_create_column(element, **kw)

@compiles(CreateIndex)
def uppercase_create_index(element, compiler, **kw):
    # 转换索引名大写
    idx = element.element
    idx.name = idx.name.upper()
    if idx.table.schema:
        idx.table.schema = idx.table.schema.upper()
    return compiler.visit_create_index(element, **kw)

把以上代码放到项目的模型初始化文件中，加载模型时会自动注册钩子，所有生成的DDL语句标识符都会统一为大写。

方案2：Alembic迁移配置适配 #

仅修改DDL生成逻辑还不够，Alembic自动生成迁移时会反射Snowflake中已有的库表结构，需要在alembic/env.py中补充配置，避免反射对比时大小写不匹配：

def run_migrations_online():
    connectable = engine_from_config(
        config.get_section(config.config_ini_section, {}),
        prefix="sqlalchemy.",
        poolclass=pool.NullPool,
    )

    with connectable.connect() as connection:
        context.configure(
            connection=connection,
            target_metadata=target_metadata,
            version_table_schema=target_metadata.schema.upper() if target_metadata.schema else None,
            # 对比对象时统一转大写后匹配，忽略大小写差异
            include_object=lambda obj, name, obj_type, reflected, compare_to: (
                name.upper() == compare_to.name.upper() if compare_to and obj_type == "table" else True
            ),
            compare_server_default=True
        )

        with context.begin_transaction():
            context.run_migrations()

注意事项 #

• 不要给标识符手动添加双引号，Snowflake中被双引号包裹的标识符会强制区分大小写，反而会破坏匹配逻辑
• 如果项目中手动指定了约束、外键名称，可参考上述编译钩子的写法补充AddConstraint、ForeignKeyConstraint的编译规则，统一转大写即可
• 如果是新项目也可以直接在声明式基类中统一做名称转换，逻辑更直观：

  from sqlalchemy.orm import DeclarativeBase
  from sqlalchemy import MetaData
  
  class Base(DeclarativeBase):
      metadata = MetaData(schema="YOUR_SCHEMA_NAME")
      def __init_subclass__(cls, **kwargs):
          super().__init_subclass__(**kwargs)
          if hasattr(cls, "__tablename__"):
              cls.__tablename__ = cls.__tablename__.upper()
  

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