python-docs-samples 实战：Cloud Run functions 中高效调用 GCP API 的全局客户端复用方案

导读

本指南以 python-docs-samples 仓库中 functions/tips-gcp-apis/ 示例为核心，讲解如何在 Cloud Run functions（原 Cloud Functions 1st gen）中通过模块级全局复用客户端库实例的方式来调用 Google Cloud API（以 Pub/Sub 为例），从而显著减少每次函数调用建立的网络连接数量。读完本文，你将掌握全局客户端复用模式的具体写法、GCP_PROJECT 环境变量的运行时差异、functions-framework 的本地调试方式，以及配套的单元测试与依赖配置，可直接照搬到其他 GCP 服务（如 Firestore、Storage）的调用场景中。

为什么要在 Cloud Functions 中复用 GCP 客户端

Cloud Run functions 是无服务器计算模型：函数实例在首次请求时被创建，随后在存活期间被复用来处理后续请求。这一特性决定了"连接开销"的优化空间：

• 若在每次函数调用内部创建新的客户端实例（例如在 gcp_api_call 函数体内执行 pubsub_v1.PublisherClient()），每次冷启动/新请求都要重新建立与 GCP API 的网络连接（TLS 握手、身份验证令牌获取等），产生不必要的延迟与资源消耗；
• 若在模块级别（即函数文件顶层）创建客户端实例，则该实例在实例整个生命周期内被所有调用共享，TCP 连接与认证信息得以复用。

这正是官方"优化网络连接"最佳实践的核心主张（即本 README 指向的 Optimizing Networking 教程 中 maintaining_persistent_connections 一节）。仓库中 functions/tips-connection-pooling/ 示例以 requests.Session() 演示了同一思想在 HTTP 调用上的应用，而 functions/tips-gcp-apis/ 则将其推广到 GCP 客户端库。

核心实现：全局 Pub/Sub 客户端复用

示例文件 main.py 的完整逻辑如下（代码以 # [START functions_tips_gcp_apis] 与 # [END functions_tips_gcp_apis] 标记，供文档引用与测试抽取）：

import os

import functions_framework
from google.cloud import pubsub_v1


# Create a global Pub/Sub client to avoid unneeded network activity
pubsub = pubsub_v1.PublisherClient()


@functions_framework.http
def gcp_api_call(request):
    project = os.getenv("GCP_PROJECT")
    request_json = request.get_json()

    topic_name = request_json["topic"]
    topic_path = pubsub.topic_path(project, topic_name)

    # Process the request
    data = b"Test message"
    pubsub.publish(topic_path, data=data)

    return "1 message published"

关键设计点逐条拆解：

1. 全局客户端实例：pubsub = pubsub_v1.PublisherClient() 位于模块顶层，函数实例被创建时执行一次，此后所有请求共用该 PublisherClient，避免不必要的网络活动。这是与"在函数内每次新建客户端"的朴素写法最本质的区别。
2. GCP_PROJECT 环境变量：project = os.getenv("GCP_PROJECT") 用于获取当前项目 ID。需要注意运行时差异——代码注释明确指出：
   • 在 Python 3.7 运行时，该变量由平台自动注入；
   • 在 Python 3.8 及以后的运行时，GCP_PROJECT 不再自动设置，需要用户在部署函数时通过环境变量手动指定（详见 functions/docs 环境变量配置页 中 Python 3.7 与 Go 1.11 的相关说明）。 因此跨运行时部署时，务必在部署命令中显式传递，例如 --set-env-vars GCP_PROJECT=your-project-id。
3. HTTP 触发与 JSON 解析：@functions_framework.http 声明这是一个 HTTP 触发的 Cloud Run functions；request.get_json() 解析请求体，并从中取出 "topic" 字段作为目标 Topic 名。
4. 发布消息：pubsub.topic_path(project, topic_name) 根据项目与 Topic 名拼接出完整资源路径（形如 projects/{project}/topics/{topic}），随后 pubsub.publish(topic_path, data=b"Test message") 将固定测试消息发布到该 Topic，函数返回字符串 "1 message published" 作为 HTTP 响应。

环境变量与运行时差异：跨版本部署注意事项

从源码注释（main.py）可以直接提取到以下事实性结论，这是该示例最有价值的"坑位提醒"：

运行时	GCP_PROJECT 来源	是否需要手动设置
Python 3.7	平台自动注入	否
Python 3.8 及以后	部署时用户自行指定	是

这意味着：如果你在较新的运行时（如 Python 3.11/3.12）部署本示例但未设置 GCP_PROJECT，os.getenv("GCP_PROJECT") 将返回 None，进而导致 topic_path 拼接出非法资源路径。部署时应通过 gcloud functions deploy 的 --set-env-vars 显式注入项目 ID，保证函数在不同运行时下行为一致。

依赖配置：客户端库与函数框架版本

requirements.txt 声明了两个运行时依赖：

• google-cloud-pubsub==2.28.0：Pub/Sub 官方 Python 客户端库，提供 PublisherClient、topic_path、publish 等核心 API；
• functions-framework==3.9.2：Cloud Run functions 的 Python 本地开发与运行框架，负责将 @functions_framework.http 装饰的函数包装成可执行的 HTTP 服务，也用于本地调试与测试启动。

两个依赖均采用精确版本锁定（==），这是 python-docs-samples 仓库的一贯风格，目的是保证示例在文档所示环境中可复现、可验证。测试环境则额外依赖 requirements-test.txt 中的 pytest 与 Flask。

本地运行与调试

在部署到云端之前，可以利用 functions-framework 在本地启动函数：

pip install -r requirements.txt
export GCP_PROJECT=your-project-id
export FUNCTIONS_TOPIC=your-topic-name
functions-framework --target gcp_api_call

然后发送 POST 请求进行验证（请求体必须为 JSON 且包含 topic 字段）：

curl -X POST -H "Content-Type: application/json" \
  -d '{"topic":"your-topic-name"}' \
  http://localhost:8080

预期响应为 1 message published。注意本地调试同样需要先设置 GCP_PROJECT，以模拟较新运行时下的生产环境变量行为。

单元测试验证：模拟请求上下文

仓库为示例提供了配套测试 main_test.py，用于验证函数在 HTTP 请求下的行为：

import json
import os

import flask
import pytest

import main


@pytest.fixture(scope="module")
def app():
    return flask.Flask(__name__)


def test_publishes_message(app):
    data = json.dumps({"topic": os.getenv("FUNCTIONS_TOPIC")})

    with app.test_request_context(data=data, content_type="application/json"):
        response = main.gcp_api_call(flask.request)
        assert response == "1 message published"

测试要点：

1. Flask 测试上下文：通过 flask.Flask(__name__) 构造一个假的"app"，再利用 test_request_context 模拟真实 HTTP 请求环境，无需真正启动服务器；
2. 请求构造：请求体 JSON 中的 topic 取值来自环境变量 FUNCTIONS_TOPIC，说明运行测试前需要预先设置该变量（本地可通过 export FUNCTIONS_TOPIC=your-topic-name 完成）；
3. 断言行为：调用 main.gcp_api_call(flask.request) 后，断言返回值恰为 "1 message published"，与函数源码的返回字符串严格一致。

运行测试命令：

pip install -r requirements-test.txt
export FUNCTIONS_TOPIC=your-topic-name
pytest main_test.py -v

模式推广：从 Pub/Sub 到其他 GCP 服务

"全局复用客户端"并非 Pub/Sub 专属，而是适用于所有提供 Python 客户端库的 GCP 服务。仓库中同系列示例可以相互印证：

• tips-connection-pooling/main.py：在模块顶层创建 requests.Session()，为普通 HTTP 出站请求提供连接池，与本文的 PublisherClient 复用是同一优化思想在不同客户端上的体现；
• 其他 tips-* 目录（如 tips-lazy-globals、tips-scopes）共同组成了官方"Cloud Functions 最佳实践"系列，分别从惰性初始化、作用域等角度处理无服务器环境下的资源生命周期问题。

实践中，将本示例中的 pubsub_v1.PublisherClient() 替换为 storage.Client()、firestore.Client() 等，即可把同样的模式应用到对象存储、文档数据库等场景。核心原则始终是：凡是需要建立网络连接、获取凭据的客户端对象，都应在模块顶层创建并复用。

部署到 Cloud Run functions

云端部署时（以 1st gen Cloud Functions 为例，参考 functions/README.md 中关于两种运行时类型的说明），在完成 认证配置 后，可使用如下命令：

gcloud functions deploy gcp_api_call \
  --runtime python311 \
  --trigger-http \
  --allow-unauthenticated \
  --set-env-vars GCP_PROJECT=your-project-id

部署成功后，调用入口 URL 即为函数触发端点，向其 POST 包含 {"topic": "..."} 的 JSON 即可触发一次 Pub/Sub 消息发布。再次强调：--set-env-vars GCP_PROJECT=... 是 Python 3.8+ 运行时下的必选项。

小结

functions/tips-gcp-apis/ 示例以最小的代码量完整演示了 Cloud Run functions 调用 GCP API 的推荐姿势：模块级全局客户端复用降低连接开销、GCP_PROJECT 环境变量的运行时差异处理、HTTP JSON 请求解析、配套单元测试与精确依赖锁定。这一模式可直接迁移到其他 GCP 服务，是无服务器场景下兼顾性能与代码简洁性的标准答案。