Cloud Functions如何获取API Gateway路由参数实现单函数多路由

核心原理 #

API Gateway转发请求到后端Cloud Function时，会透传原始请求的路径信息，你只需要在函数中读取请求对应的路径标识，做一层映射判断即可切换BigQuery写入目标，不需要为每个路径单独部署函数。

操作步骤 #

• 调整API Gateway路由配置
  保留原有/user、/client、/order三条路径的匹配规则，将三条规则的后端服务统一指向你要复用的Python Cloud Function触发地址，路径翻译策略保持默认即可，不需要额外修改API规范的其他配置。
• 编写Cloud Function的路径判断与分发逻辑
  Python运行时的HTTP触发Cloud Function基于Flask实现，你既可以直接从请求对象读取路径，也可以从API Gateway自动注入的转发头中获取用户实际请求的原始路径，避免路径改写导致的判断错误。参考实现代码如下：

import functions_framework
from google.cloud import bigquery

# 全局初始化BigQuery客户端，复用连接降低冷启动开销
bq_client = bigquery.Client()
# 维护路径与BigQuery目标表的映射关系，新增业务场景只需要改这里
ROUTE_TABLE_MAPPING = {
    "/user": "替换为你的GCP项目ID.替换为数据集名.user",
    "/client": "替换为你的GCP项目ID.替换为数据集名.client",
    "/order": "替换为你的GCP项目ID.替换为数据集名.order"
}

@functions_framework.http
def bq_write_handler(request):
    # 优先读取转发头中的原始请求路径，兼容路径前缀剥离配置；兼容尾部斜杠请求
    request_path = request.headers.get("X-Forwarded-Path", request.path).rstrip("/")
    
    # 非法路径直接返回404
    if request_path not in ROUTE_TABLE_MAPPING:
        return ("Unsupported path", 404)
    
    # 兼容不同请求方法的参数解析
    req_payload = {}
    if request.method == "GET":
        req_payload = dict(request.args)
    else:
        if request.is_json:
            req_payload = request.get_json(silent=True) or {}
        elif request.form:
            req_payload = dict(request.form)
    
    # 写入对应目标表
    target_table = ROUTE_TABLE_MAPPING[request_path]
    insert_err = bq_client.insert_rows_json(target_table, [req_payload])
    if insert_err:
        return (f"Insert failed: {insert_err}", 500)
    
    return ("Success", 200)

优化注意点 #

• 新增业务路由时，只需要在API Gateway添加对应路径规则指向该函数，同时在ROUTE_TABLE_MAPPING中新增路径和表的映射即可，无需重复部署函数。
• BigQuery客户端必须放在请求处理函数外做全局初始化，不要在每次请求进来时重新创建客户端，能显著降低冷启动耗时、减少接口响应延迟。
• 路径处理加rstrip("/")是为了兼容用户请求带尾部斜杠的场景，比如/user/也能正常匹配路由，不会误返回404。
• 优先读取X-Forwarded-Path头是为了兼容你后续如果配置API网关路径前缀剥离的场景，该头由API Gateway自动写入，存储的是用户实际访问的原始路径，比直接读request.path更稳妥。

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