如何通过Google Connector将Alembic连接至Cloud SQL？

直接修改Alembic的env.py文件即可实现，无需在alembic.ini中配置sqlalchemy.url，具体操作如下：

1. 修改alembic/env.py #

打开项目中的alembic/env.py文件，替换默认的引擎创建逻辑，复用你FastAPI项目里的连接池初始化代码：

from logging.config import fileConfig

from sqlalchemy import create_engine
from sqlalchemy.engine import Engine
from alembic import context
from google.cloud.sql.connector import Connector
# 导入你的FastAPI项目中的Base模型（根据实际文件路径调整）
from app.models import Base

# 保留Alembic默认配置对象
config = context.config

# 保留默认日志配置
if config.config_file_name is not None:
    fileConfig(config.config_file_name)

# 关联模型元数据，用于生成迁移脚本
target_metadata = Base.metadata

# 复用FastAPI中的连接池初始化逻辑
def init_connection_pool(connector: Connector) -> Engine:
    def getconn():
        conn = connector.connect(
            "project:region:instance-name",  # Cloud SQL实例连接名
            "pg8000",
            user="my-user",
            password="my-password",
            db="my-database",
            ip_type="public"  # 改为"private"可使用私有IP连接
        )
        return conn

    SQLALCHEMY_DATABASE_URL = "postgresql+pg8000://"
    engine = create_engine(SQLALCHEMY_DATABASE_URL, creator=getconn)
    return engine

def run_migrations_offline() -> None:
    """离线迁移模式（可忽略，我们使用在线模式）"""
    url = config.get_main_option("sqlalchemy.url")
    context.configure(
        url=url,
        target_metadata=target_metadata,
        literal_binds=True,
        dialect_opts={"paramstyle": "named"},
    )

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

def run_migrations_online() -> None:
    """在线迁移模式，使用Google Connector创建数据库连接"""
    connector = Connector()
    connectable = init_connection_pool(connector)

    with connectable.connect() as connection:
        context.configure(
            connection=connection, target_metadata=target_metadata
        )

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

# 自动判断运行模式，默认执行在线迁移
if context.is_offline_mode():
    run_migrations_offline()
else:
    run_migrations_online()

2. 关键注意事项 #

• 确保from app.models import Base的路径正确，指向你FastAPI项目中定义Base = declarative_base()的文件。
• 将代码中的project:region:instance-name、my-user、my-password、my-database替换为你的Cloud SQL实际信息。
• 无需修改alembic.ini中的sqlalchemy.url，因为我们已在env.py中自定义了引擎创建逻辑。

关于Cloud SQL Proxy的补充说明 #

你提到运行cloud-sql-proxy PROJECT_NAME:REGION:SQLDB时项目可正常工作，这种场景下也可以直接在alembic.ini中配置本地代理的连接URL：

sqlalchemy.url = postgresql+pg8000://my-user:my-password@127.0.0.1:5432/my-database

但这种方式依赖代理进程持续运行，而修改env.py的方式可直接通过Google Connector连接，无需额外启动代理服务。

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