如何通过Alembic连接Google Cloud SQL执行FastAPI应用迁移？

以下是两种经过验证的方案，涵盖本地开发和无代理生产场景，附带完整示例代码解决你使用Google Cloud Python Connector失败的问题。

前置准备 #

• 已创建Cloud SQL实例（以PostgreSQL为例，MySQL可替换对应驱动）
• 安装依赖：

  pip install alembic sqlalchemy google-cloud-sql-connector pg8000  # PostgreSQL
  # 若为MySQL，替换pg8000为pymysql
  

• 配置GCP服务账号（授予Cloud SQL Client角色），并设置环境变量：

  export GOOGLE_APPLICATION_CREDENTIALS="/path/to/your/service-account-key.json"
  

方案1：Cloud SQL Auth Proxy（本地开发首选） #

此方案无需修改代码，通过代理转发流量，适合本地调试。

1. 下载并启动Cloud SQL Auth Proxy：

   # Linux/macOS
   ./cloud-sql-proxy your-gcp-project:your-region:your-sql-instance-name
   # Windows
   cloud-sql-proxy.exe your-gcp-project:your-region:your-sql-instance-name
   

2. 修改alembic.ini中的数据库连接URL：

   sqlalchemy.url = postgresql://db_username:db_password@127.0.0.1:5432/db_name
   

3. 执行迁移命令：

   alembic upgrade head
   

方案2：Google Cloud Python Connector（无代理生产场景） #

通过修改Alembic的env.py实现直接连接，解决你之前的失败问题。

1. 编辑alembic/env.py，替换原有连接逻辑：

   import os
   from google.cloud.sql.connector import Connector, IPTypes
   import sqlalchemy
   from your_fastapi_app.models import Base  # 导入你的FastAPI模型Base类
   
   def create_sqlalchemy_engine():
       # 初始化Connector
       connector = Connector()
   
       def get_connection():
           return connector.connect(
               "your-gcp-project:your-region:your-sql-instance-name",
               "pg8000",  # PostgreSQL用pg8000，MySQL替换为"pymysql"
               user="db_username",
               password="db_password",
               db="db_name",
               # 优先使用私有IP（若运行在GCP内部），否则用公网IP
               ip_type=IPTypes.PRIVATE if os.getenv("USE_PRIVATE_IP") else IPTypes.PUBLIC
           )
   
       # 创建SQLAlchemy引擎
       return sqlalchemy.create_engine(
           "postgresql+pg8000://",  # MySQL对应"mysql+pymysql://"
           creator=get_connection,
           pool_recycle=3600  # 避免连接超时
       )
   
   # 替换原有target_metadata和connectable定义
   target_metadata = Base.metadata
   connectable = create_sqlalchemy_engine()
   
   # 保留原有run_migrations_offline和run_migrations_online函数，确保使用connectable
   def run_migrations_offline():
       """Run migrations in 'offline' mode.
   
       This configures the context with just a URL
       and not an Engine, though an Engine is acceptable
       here as well.  By skipping the Engine creation
       we don't even need a DBAPI to be available.
   
       Calls to context.execute() here emit the given string to the
       script output.
   
       """
       url = connectable.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():
       """Run migrations in 'online' mode.
   
       In this scenario we need to create an Engine
       and associate a connection with the context.
   
       """
       with connectable.connect() as connection:
           context.configure(
               connection=connection, target_metadata=target_metadata
           )
   
           with context.begin_transaction():
               context.run_migrations()
   

2. 执行迁移命令：

   alembic upgrade head
   

关键注意事项 #

• 确保服务账号拥有Cloud SQL Client角色权限
• 使用私有IP时，需确保运行环境（如GCE/GKE）与Cloud SQL实例在同一VPC
• 生产环境建议使用私有IP+服务账号自动认证（无需手动设置密钥文件）

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