如何让Alembic_utils的create_entity支持if not exists创建扩展？

我正在从Liquibase迁移到Alembic，需要兼容两种数据库环境创建uuid-ossp扩展：

• 已有数据库：由Liquibase创建Schema，已预装uuid-ossp扩展
• 全新数据库：基于Alembic搭建，未安装该扩展

目前用Alembic的op.create_entity(PGExtension(...))创建扩展，但没有类似建表时的if_not_exists选项；且SQLAlchemy反射机制无法获取已安装扩展列表，没法通过反射判断扩展是否存在。我需要实现「仅在扩展不存在时执行创建」的逻辑，求可行思路。

现有代码示例：

def upgrade() -> None:
    # ### commands auto generated by Alembic - please adjust! ###
    public_uuid_ossp = PGExtension(
        schema="public",
        signature="uuid-ossp"
    )
    op.create_entity(public_uuid_ossp)

方法1：直接执行带IF NOT EXISTS的原生SQL #

PostgreSQL原生支持创建扩展时添加IF NOT EXISTS关键字，直接用op.execute执行SQL是最简洁的方式：

def upgrade() -> None:
    op.execute("CREATE EXTENSION IF NOT EXISTS \"uuid-ossp\" WITH SCHEMA public;")

方法2：查询系统表检查状态后再创建 #

通过查询PostgreSQL的pg_extension系统表，判断目标扩展是否存在，再决定是否调用create_entity：

from alembic import op
from sqlalchemy import text

def upgrade() -> None:
    conn = op.get_bind()
    # 查询public schema下的uuid-ossp扩展是否存在
    result = conn.execute(
        text("SELECT 1 FROM pg_extension WHERE extname = 'uuid-ossp' AND extnamespace = (SELECT oid FROM pg_namespace WHERE nspname = 'public')")
    ).fetchone()
    
    if not result:
        public_uuid_ossp = PGExtension(
            schema="public",
            signature="uuid-ossp"
        )
        op.create_entity(public_uuid_ossp)

方法3：自定义封装复用逻辑 #

如果需要在多个迁移脚本中复用该逻辑，可以封装一个自定义操作函数：

from alembic import op
from sqlalchemy import text
from alembic.operations import Operations
from alembic.postgresql import PGExtension

@Operations.register_operation("create_extension_if_not_exists")
def create_extension_if_not_exists(operations, extension):
    conn = operations.get_bind()
    query = text(
        "SELECT 1 FROM pg_extension "
        "WHERE extname = :extname AND extnamespace = (SELECT oid FROM pg_namespace WHERE nspname = :schema)"
    )
    result = conn.execute(query, {"extname": extension.signature, "schema": extension.schema}).fetchone()
    
    if not result:
        operations.create_entity(extension)

# 使用示例
def upgrade() -> None:
    public_uuid_ossp = PGExtension(
        schema="public",
        signature="uuid-ossp"
    )
    op.create_extension_if_not_exists(public_uuid_ossp)

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