)
示例工程【免费下载链接】python-docs-samplesCode samples used on cloud.google.com项目地址https://gitcode.com/GitHub_Trending/py/python-docs-samples点击查看免费下载本文以 Google Cloud 官方示例仓库 python-docs-samples 中的 SQL Server 连接演示应用 为核心系统讲解两种连接 Cloud SQL for SQL Server 的方式——推荐的 Cloud SQL Python Connector 与 Cloud SQL Auth ProxyTCP 直连并覆盖本地运行、App EngineStandard/Flexible、Cloud Run、Cloud Functions 四种部署场景。读完本文你将掌握基于 SQLAlchemy 连接池的完整配置方法、环境变量体系与源码级连接原理可直接复用到自己的生产项目中。项目概览一个Tabs vs Spaces投票应用的连接骨架示例应用是一个基于 Flask 的投票 Web 应用提供GET /展示票数与最近投票和POST /votes投出 TABS 或 SPACES 一票两个接口页面模板位于 templates/index.html。其核心价值在于演示了在 Python 中连接 Cloud SQL 实例的两种官方推荐路径Cloud SQL Python Connector推荐以原生 Python 包形式提供与 Auth Proxy 相同的功能无需额外进程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提示。这套按环境变量自动分流的设计让你可以用同一份代码轻松适配本地开发、代理连接、托管平台等多种运行环境。前置准备实例、数据库与凭据在运行示例前需要完成以下四步基础工作配置 Python 开发环境并创建 Google Cloud 项目创建 2nd Gen Cloud SQL 实例记录连接字符串、数据库用户名和密码为应用创建数据库记录数据库名创建具备Cloud SQL Client权限的服务账号并下载 JSON 密钥用于认证[连接方式一] 使用 ADC 默认凭据读取。其中INSTANCE_CONNECTION_NAME可在云控制台实例的 Overview 页面查看也可通过命令获取gcloud sql instances describe INSTANCE_NAME --formatvalue(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_typeip_type, refresh_strategyLAZY)几个要点PRIVATE_IP环境变量设置后走私有 IP否则默认使用公网 IPIPTypes.PUBLIC是内网/VPC 部署场景的关键开关refresh_strategyLAZY连接元数据按需刷新避免轮询开销DB_ROOT_CERT可选配置当实例要求 SSL 时需下载实例 CA 证书并设置cafile与validate_hostFalse——这是官方已知问题的一个 workaround两种连接方式下均适用。自定义连接工厂与连接池Connector 通过getconn()工厂函数产出pytds.Connection再交给 SQLAlchemy 的create_engine(..., creatorgetconn)创建连接池pool sqlalchemy.create_engine( mssqlpytds://, creatorgetconn, pool_size5, # 常驻连接上限 max_overflow2, # 允许临时超出的连接数 pool_timeout30, # 获取连接的等待上限秒 pool_recycle1800, # 连接最长存活 30 分钟到期重建 )注意此时 URL 是mssqlpytds://不含主机信息实际连接参数全部由getconn()提供。测试代码 connection_test.py 中也有断言str(app.db.url) mssqlpytds://印证了这一设计。连接方式二TCP 直连配合 Cloud SQL Auth Proxy当使用 Cloud SQL Auth Proxy或部署到已配置cloud_sql_instances的 App Engine Flex时应用把代理监听的本地地址当作普通数据库主机通过 TCP 建立连接。核心实现见 connect_tcp.pydb_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( drivernamemssqlpytds, usernamedb_user, passworddb_pass, databasedb_name, hostdb_host, portdb_port, ), connect_argsconnect_args, # SSL 证书配置DB_ROOT_CERT 时启用 pool_size5, max_overflow2, pool_timeout30, pool_recycle1800, )这段代码的等价 URL 写法是mssqlpytds://db_user:db_passdb_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_idInteger 主键、time_castDateTime、candidateString(6)三列并create_all()建表注释中特别说明惰性初始化主要是为了便于测试Cloud Run / App Engine 场景可改为加载时立即初始化。参数化查询防注入投票写入使用 SQLAlchemy 的text()参数化语句app.pystmt 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 / macOSexport GOOGLE_APPLICATION_CREDENTIALS/path/to/service/account/key.json export INSTANCE_CONNECTION_NAMEPROJECT_ID:INSTANCE_REGION:INSTANCE_NAME export DB_USERYOUR_DB_USER_NAME export DB_PASSYOUR_DB_PASSWORD export DB_NAMEYOUR_DB_NAMEWindows / PowerShell$env:GOOGLE_APPLICATION_CREDENTIALS/path/to/service/account/key.json $env:INSTANCE_CONNECTION_NAMEPROJECT_ID:INSTANCE_REGION:INSTANCE_NAME $env:DB_USERYOUR_DB_USER_NAME $env:DB_PASSYOUR_DB_PASSWORD $env:DB_NAMEYOUR_DB_NAME说明GOOGLE_APPLICATION_CREDENTIALS指向服务账号 JSON 密钥路径供 ADCApplication 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 FlexibleFlexible 环境使用runtime: customenv: flex且通常配合实例侧的代理自动注入因此连接方式切换为 TCP 直连app.flexible.yamlruntime: 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_NAMEtcp:1433这里INSTANCE_HOST172.17.0.1是 GAE Flex 容器内访问宿主机的固定地址cloud_sql_instances声明实例并通过tcp:1433暴露到本机端口——这就是平台帮你起代理应用只管 TCP 连的典型形态。部署命令为gcloud app deploy app.flexible.yaml。同样私有 IP 场景可追加PRIVATE_IP: True。部署到 Cloud RunCloud Run 场景推荐直接使用 Python Connector无需代理进程部署命令gcloud run deploy cloud-sql-demo \ --allow-unauthenticated \ --set-env-vars INSTANCE_CONNECTION_NAMEPROJECT_ID:INSTANCE_REGION:INSTANCE_NAME \ --set-env-vars DB_USERYOUR_DB_USER_NAME \ --set-env-vars DB_PASSYOUR_DB_PASSWORD \ --set-env-vars DB_NAMEYOUR_DB_NAME部署完成后浏览器访问输出末尾的 URL 即可验证。私有 IP 场景需追加--set-env-vars PRIVATE_IPTrue并指定--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 FunctionsGen2示例同时提供了 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_NAMEPROJECT_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_IPTrue与--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 APIv1beta4的connectSettings接口动态下载实例服务端 CA 证书到 certs/ca.pem 并设置DB_ROOT_CERT测试用例覆盖GET /返回 200 且页面含 Tabs VS SpacesPOST /votesteamSPACES返回 200 且含成功文案test_connector_connection移除INSTANCE_HOST后验证走 Connector 路径且引擎 URL 为mssqlpytds://。这组测试同时验证了两种连接方式都能真实读写数据库是示例可运行、可验证的直接证据。常见问题与生产实践小结连接池参数怎么调默认pool_size5, max_overflow2, pool_timeout30, pool_recycle1800。应用的总并发连接上限约为pool_size max_overflowpool_recycle1800可规避长连接被数据库端回收导致的失效高并发场景应结合 Cloud SQL 实例的最大连接数SQL Server 受实例内存规格影响合理上调而非盲目加大。SSL 证书报错需要 SSL 的实例请下载服务端 CA 证书设置DB_ROOT_CERT并传入cafile与validate_hostFalse已知问题 workaround两种连接方式均适用。私有 IP 怎么连在连接代码识别PRIVATE_IP环境变量对应IPTypes.PRIVATE各部署平台分别通过env_variables、--set-env-vars或PRIVATE_IPTrue注入并在 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 应用中。赞分享示例工程【免费下载链接】python-docs-samplesCode samples used on cloud.google.com项目地址https://gitcode.com/GitHub_Trending/py/python-docs-samples点击查看免费下载相关推荐Python 连接 SQL Server 实战sql-server-samples 中 pyodbc 与 pymssql 连接样例全解析Python 连接 SQL Server 实战sql server samples 中 pyodbc 与 pymssql 连接样例全解析 本文围绕 sql s示例工程数据库教程后端解锁隐藏功能用GalaxyBudsClient掌控你的三星耳机解锁隐藏功能用GalaxyBudsClient掌控你的三星耳机 GalaxyBudsClient是一款功能强大的三星Galaxy Buds耳机管理工具支持W桌面应用智能硬件蓝牙如何快速搭好自托管游戏串流服务器Sunshine 入门指南如何快速搭好自托管游戏串流服务器Sunshine 入门指南 Sunshine 是一个自托管游戏串流服务器专为 Moonlight 客户端设计。假设你出差只带音视频后端上一篇ComfyUI-KJNodes从节点连线到思维导图AI工作流的革命性进化下一篇如何在Windows 10上实现Android应用原生运行3步完成逆向工程移植创作声明:本文部分内容由AI辅助生成(AIGC),仅供参考