使用 Python Connector 与 SQLAlchemy 连接 Cloud SQL for SQL Server 全指南（python-docs-samples）

• 示例工程

本文以 Google Cloud 官方示例仓库 python-docs-samples 中的 SQL Server 连接演示应用 为核心，系统讲解两种连接 Cloud SQL for SQL Server 的方式——推荐的 Cloud SQL Python Connector 与 Cloud SQL Auth Proxy（TCP 直连），并覆盖本地运行、App Engine（Standard/Flexible）、Cloud Run、Cloud Functions 四种部署场景。读完本文，你将掌握基于 SQLAlchemy 连接池的完整配置方法、环境变量体系与源码级连接原理，可直接复用到自己的生产项目中。

项目概览：一个"Tabs vs Spaces"投票应用的连接骨架

示例应用是一个基于 Flask 的投票 Web 应用，提供GET /（展示票数与最近投票）和POST /votes（投出 TABS 或 SPACES 一票）两个接口，页面模板位于 templates/index.html。其核心价值在于演示了在 Python 中连接 Cloud SQL 实例的两种官方推荐路径：

1. Cloud SQL Python Connector（推荐）：以原生 Python 包形式提供与 Auth Proxy 相同的功能，无需额外进程；
2. Cloud SQL Auth Proxy（即 TCP 直连模式）：通过代理进程建立加密隧道后，应用以普通 TCP 方式访问本地地址。

应用在启动时根据环境变量自动选择连接方式。在 app.py 的init_connection_pool()中可以看到选择逻辑：

• 若定义了INSTANCE_HOST（如127.0.0.1），调用connect_tcp_socket()走 TCP 直连；
• 若定义了INSTANCE_CONNECTION_NAME（格式project:region:instance），调用connect_with_connector()走 Python Connector；
• 两者都未定义则抛出ValueError提示。

这套"按环境变量自动分流"的设计，让你可以用同一份代码轻松适配本地开发、代理连接、托管平台等多种运行环境。

前置准备：实例、数据库与凭据

在运行示例前，需要完成以下四步基础工作：

1. 配置 Python 开发环境并创建 Google Cloud 项目；
2. 创建 2nd Gen Cloud SQL 实例，记录连接字符串、数据库用户名和密码；
3. 为应用创建数据库，记录数据库名；
4. 创建具备Cloud SQL Client权限的服务账号，并下载 JSON 密钥用于认证（[连接方式一] 使用 ADC 默认凭据读取）。

其中INSTANCE_CONNECTION_NAME可在云控制台实例的 Overview 页面查看，也可通过命令获取：

gcloud sql instances describe <INSTANCE_NAME> --format='value(connectionName)'

它会输出类似PROJECT_ID:REGION:INSTANCE_NAME的三段式连接名，这是 Python Connector 连接实例的关键参数。

连接方式一：Cloud SQL Python Connector（推荐）

该方式将认证、加密、连接管理全部封装在 Python 进程内部，无需在本地额外启动代理进程。核心实现见 connect_connector.py，依赖cloud-sql-python-connector与python-tds（详见 requirements.txt）。

环境变量与初始化流程

instance_connection_name = os.environ["INSTANCE_CONNECTION_NAME"] # 'project:region:instance' db_user = os.environ.get("DB_USER", "") # 'my-db-user' db_pass = os.environ["DB_PASS"] # 'my-db-password' db_name = os.environ["DB_NAME"] # 'my-database' ip_type = IPTypes.PRIVATE if os.environ.get("PRIVATE_IP") else IPTypes.PUBLIC connector = Connector(ip_type=ip_type, refresh_strategy="LAZY")

几个要点：

• PRIVATE_IP环境变量：设置后走私有 IP，否则默认使用公网 IP（IPTypes.PUBLIC），是内网/VPC 部署场景的关键开关；
• refresh_strategy="LAZY"：连接元数据按需刷新，避免轮询开销；
• DB_ROOT_CERT可选配置：当实例要求 SSL 时，需下载实例 CA 证书，并设置cafile与validate_host=False——这是官方已知问题的一个 workaround，两种连接方式下均适用。

自定义连接工厂与连接池

Connector 通过getconn()工厂函数产出pytds.Connection，再交给 SQLAlchemy 的create_engine(..., creator=getconn)创建连接池：

pool = sqlalchemy.create_engine( "mssql+pytds://", creator=getconn, pool_size=5, # 常驻连接上限 max_overflow=2, # 允许临时超出的连接数 pool_timeout=30, # 获取连接的等待上限（秒） pool_recycle=1800, # 连接最长存活 30 分钟，到期重建 )

注意此时 URL 是mssql+pytds://（不含主机信息），实际连接参数全部由getconn()提供。测试代码 connection_test.py 中也有断言str(app.db.url) == "mssql+pytds://"，印证了这一设计。

连接方式二：TCP 直连（配合 Cloud SQL Auth Proxy）

当使用 Cloud SQL Auth Proxy（或部署到已配置cloud_sql_instances的 App Engine Flex）时，应用把代理监听的本地地址当作普通数据库主机，通过 TCP 建立连接。核心实现见 connect_tcp.py：

db_host = os.environ["INSTANCE_HOST"] # 本地代理地址，如 '127.0.0.1' db_user = os.environ["DB_USER"] db_pass = os.environ["DB_PASS"] db_name = os.environ["DB_NAME"] db_port = os.environ["DB_PORT"] # SQL Server 默认 1433 pool = sqlalchemy.create_engine( sqlalchemy.engine.url.URL.create( drivername="mssql+pytds", username=db_user, password=db_pass, database=db_name, host=db_host, port=db_port, ), connect_args=connect_args, # SSL 证书配置（DB_ROOT_CERT 时启用） pool_size=5, max_overflow=2, pool_timeout=30, pool_recycle=1800, )

这段代码的等价 URL 写法是mssql+pytds://<db_user>:<db_pass>@<db_host>:<db_port>/<db_name>。与 Connector 方式相比，它显式携带主机与端口；同样支持DB_ROOT_CERT的 SSL workaround。连接池参数与方式一完全一致，说明无论走哪种通道，连接池的伸缩策略都是统一的，这是生产中保证连接数量可控的关键。

应用源码解析：连接池初始化、建表与安全查询

惰性初始化与建表

app.py 中，连接池db全局变量初始为None，通过@app.before_request钩子在首个请求到来时惰性初始化并调用migrate_db()：

• migrate_db()先用sqlalchemy.inspect()检查votes表是否存在，不存在则以Table(...)定义vote_id（Integer 主键）、time_cast（DateTime）、candidate（String(6)）三列并create_all()建表；
• 注释中特别说明：惰性初始化主要是为了便于测试，Cloud Run / App Engine 场景可改为加载时立即初始化。

参数化查询防注入

投票写入使用 SQLAlchemy 的text()参数化语句（app.py）：

stmt = sqlalchemy.text( "INSERT INTO votes (time_cast, candidate) VALUES (:time_cast, :candidate)" ) with db.connect() as conn: conn.execute(stmt, parameters={"time_cast": time_cast, "candidate": team}) conn.commit()

with db.connect()保证连接在语句结束后（无论是否出错）归还连接池；用命名占位符替代字符串拼接，可有效防御 SQL 注入。读取侧（get_index_context）同样使用:candidate参数化统计票数，并注意 SQL Server 语法SELECT TOP(5) ... ORDER BY time_cast DESC与标准 SQL 的差异。

本地运行：环境变量与虚拟环境

Linux / macOS

export GOOGLE_APPLICATION_CREDENTIALS='/path/to/service/account/key.json' export INSTANCE_CONNECTION_NAME='<PROJECT_ID>:<INSTANCE_REGION>:<INSTANCE_NAME>' export DB_USER='<YOUR_DB_USER_NAME>' export DB_PASS='<YOUR_DB_PASSWORD>' export DB_NAME='<YOUR_DB_NAME>'

Windows / PowerShell

$env:GOOGLE_APPLICATION_CREDENTIALS="/path/to/service/account/key.json" $env:INSTANCE_CONNECTION_NAME="<PROJECT_ID>:<INSTANCE_REGION>:<INSTANCE_NAME>" $env:DB_USER="<YOUR_DB_USER_NAME>" $env:DB_PASS="<YOUR_DB_PASSWORD>" $env:DB_NAME="<YOUR_DB_NAME>"

说明：GOOGLE_APPLICATION_CREDENTIALS指向服务账号 JSON 密钥路径，供 ADC（Application Default Credentials）完成认证。本地直连（不走 Connector）时，可改用INSTANCE_HOST指向本地代理端口并补DB_PORT（默认 1433）。

安装依赖并启动

virtualenv --python python3 env source env/bin/activate pip install -r requirements.txt python app.py

启动后访问http://127.0.0.1:8080即可看到投票页面并实际读写数据库。虚拟环境隔离是避免依赖冲突（Flask、SQLAlchemy、python-tds、cloud-sql-python-connector 等）的推荐做法。

部署到 App Engine Standard

将 app.standard.yaml 中<...>占位符替换为真实值后执行gcloud app deploy app.standard.yaml。标准版 YAML 的关键配置：

runtime: python310 entrypoint: gunicorn -b :$PORT app:app env_variables: INSTANCE_CONNECTION_NAME: <PROJECT_ID>:<INSTANCE_REGION>:<INSTANCE_NAME> DB_USER: <YOUR_DB_USER_NAME> DB_PASS: <YOUR_DB_PASSWORD> DB_NAME: <YOUR_DB_NAME>

若要通过私有 IP连接，还需在env_variables中追加PRIVATE_IP: True（对应源码中IPTypes.PRIVATE分支）。

部署到 App Engine Flexible

Flexible 环境使用runtime: custom+env: flex，且通常配合实例侧的代理自动注入，因此连接方式切换为 TCP 直连（app.flexible.yaml）：

runtime: custom env: flex entrypoint: gunicorn -b :$PORT app:app env_variables: INSTANCE_HOST: '172.17.0.1' DB_PORT: '1433' DB_USER: <YOUR_DB_USER_NAME> DB_PASS: <YOUR_DB_PASSWORD> DB_NAME: <YOUR_DB_NAME> beta_settings: cloud_sql_instances: <PROJECT_ID>:<INSTANCE_REGION>:<INSTANCE_NAME>=tcp:1433

这里INSTANCE_HOST=172.17.0.1是 GAE Flex 容器内访问宿主机的固定地址，cloud_sql_instances声明实例并通过tcp:1433暴露到本机端口——这就是"平台帮你起代理，应用只管 TCP 连"的典型形态。部署命令为gcloud app deploy app.flexible.yaml。同样，私有 IP 场景可追加PRIVATE_IP: True。

部署到 Cloud Run

Cloud Run 场景推荐直接使用 Python Connector（无需代理进程），部署命令：

gcloud run deploy cloud-sql-demo \ --allow-unauthenticated \ --set-env-vars INSTANCE_CONNECTION_NAME='<PROJECT_ID>:<INSTANCE_REGION>:<INSTANCE_NAME>' \ --set-env-vars DB_USER='<YOUR_DB_USER_NAME>' \ --set-env-vars DB_PASS='<YOUR_DB_PASSWORD>' \ --set-env-vars DB_NAME='<YOUR_DB_NAME>'

部署完成后浏览器访问输出末尾的 URL 即可验证。私有 IP 场景需追加--set-env-vars PRIVATE_IP=True并指定--vpc-connector <YOUR_VPC_CONNECTOR>。

推荐：用 Secret Manager 注入凭据

官方建议不要用环境变量承载数据库密码，而是启用 Cloud Run 的 Secret Manager 集成，运行时将 Secret 注入为环境变量：

# 先创建 Secret echo -n $INSTANCE_CONNECTION_NAME | \ gcloud secrets create [INSTANCE_CONNECTION_NAME_SECRET] --data-file=- # 再绑定 Secret 到环境变量部署 gcloud run deploy cloud-sql-demo \ --allow-unauthenticated \ --update-secrets INSTANCE_CONNECTION_NAME=[INSTANCE_CONNECTION_NAME_SECRET]:latest,\ DB_USER=[DB_USER_SECRET]:latest, \ DB_PASS=[DB_PASS_SECRET]:latest, \ DB_NAME=[DB_NAME_SECRET]:latest

[SECRET]:latest表示使用 Secret 的 latest 版本；服务会在运行时把 Secret 值注入到对应环境变量，应用代码无需任何改动。这也是原文档中反复强调的密码安全实践——环境变量虽便捷但不安全，生产环境务必迁移到 Secret Manager。

部署到 Cloud Functions（Gen2）

示例同时提供了 main.py 的 Cloud Functions 入口：它复用app.py中的init_connection_pool/migrate_db/get_index_context/save_vote，以functions_framework.http暴露 HTTP 函数，并在请求上下文中完成连接池的惰性初始化。部署命令：

gcloud functions deploy votes --gen2 --runtime python310 --trigger-http \ --allow-unauthenticated \ --entry-point votes \ --region <INSTANCE_REGION> \ --set-env-vars INSTANCE_CONNECTION_NAME=<PROJECT_ID>:<INSTANCE_REGION>:<INSTANCE_NAME> \ --set-env-vars DB_USER=$DB_USER \ --set-env-vars DB_PASS=$DB_PASS \ --set-env-vars DB_NAME=$DB_NAME

注意--entry-point votes对应 main.py 中@functions_framework.http修饰的votes函数。私有 IP 场景追加--set-env-vars PRIVATE_IP=True与--vpc-connector <YOUR_VPC_CONNECTOR>。部署完成后同样以输出末尾的 URL 访问。

自动化测试：如何验证连接与功能

仓库自带的 connection_test.py 展示了完整的验证思路，可作为你接入 CI 的参考：

• setup_test_env()把SQLSERVER_*系列测试环境变量映射为应用所需的DB_USER/DB_PASS/DB_NAME/DB_PORT/INSTANCE_HOST/INSTANCE_CONNECTION_NAME，并通过 SQL Admin API（v1beta4的connectSettings接口）动态下载实例服务端 CA 证书到 certs/ca.pem 并设置DB_ROOT_CERT；
• 测试用例覆盖：GET /返回 200 且页面含 "Tabs VS Spaces"；POST /votes（team=SPACES）返回 200 且含成功文案；test_connector_connection移除INSTANCE_HOST后验证走 Connector 路径且引擎 URL 为mssql+pytds://。

这组测试同时验证了两种连接方式都能真实读写数据库，是示例"可运行、可验证"的直接证据。

常见问题与生产实践小结

• 连接池参数怎么调？默认pool_size=5, max_overflow=2, pool_timeout=30, pool_recycle=1800。应用的总并发连接上限约为pool_size + max_overflow；pool_recycle=1800可规避长连接被数据库端回收导致的失效；高并发场景应结合 Cloud SQL 实例的最大连接数（SQL Server 受实例内存规格影响）合理上调，而非盲目加大。
• SSL 证书报错？需要 SSL 的实例请下载服务端 CA 证书，设置DB_ROOT_CERT并传入cafile与validate_host=False（已知问题 workaround），两种连接方式均适用。
• 私有 IP 怎么连？在连接代码识别PRIVATE_IP环境变量（对应IPTypes.PRIVATE）；各部署平台分别通过env_variables、--set-env-vars或PRIVATE_IP=True注入，并在 App Engine Flex / Cloud Run 上配置对应的 VPC 能力。
• 密钥安全：所有部署场景都应优先使用 Secret Manager 管理DB_PASS等敏感信息，避免把凭据硬编码在 YAML 或命令行中。

从 connect_connector.py 到 connect_tcp.py，再到 app.py 与 main.py，本示例完整覆盖了"连接初始化 → 建表 → 安全读写 → 多平台部署"的闭环。你只需替换环境变量与部署命令中的占位符，即可将其移植到任何连接 Cloud SQL for SQL Server 的 Python 应用中。

• 示例工程