尧图网站建设 
示例工程【免费下载链接】python-docs-samplesCode samples used on cloud.google.com项目地址https://gitcode.com/GitHub_Trending/py/python-docs-samples点击查看免费下载导读本指南围绕 eventarc/README.md 展开介绍如何用 Flask 编写接收 CloudEvents 事件的服务端应用并部署到 Cloud Run用于响应 Pub/Sub 消息、Cloud Storage 审计日志等事件。读完本文你将掌握 Pub/Sub、审计日志Audit Logs与通用Generic三类事件接收器的核心实现、本地运行、自动化测试与容器化部署的完整实战方案。Eventarc 是什么Eventarc 是 Google Cloud 上统一的事件路由服务它将来自不同事件源如 Cloud Pub/Sub、Cloud Storage、IAM 审计日志等的事件统一封装为 CloudEvents 格式并投递到事件接收端点通常是一个 Cloud Run 服务。README 明确说明该目录包含用于 Eventarc 的 Python 示例这些示例以 Flask 为基础聚焦于接收事件并做出响应这一核心职责。示例全景README 将本目录的示例划分为三类对应 Eventarc 最常见的三种接入场景示例目录仓库根路径事件类型核心处理逻辑Pub/Subeventarc/pubsub/Cloud Pub/Sub 消息解码消息 payload返回问候语并携带事件 IDAudit Logs – Cloud Storageeventarc/audit-storage/Cloud Storage 写操作审计日志从 CloudEvent 中提取存储桶 subject 并打印Genericeventarc/generic/任意 HTTP 触发的 CloudEvent原样回显事件请求头与请求体此外仓库中还包含两个进阶变体示例audit_iamIAM 审计日志与storage_handlerCloud Storage CloudEvent 直连下文将一并结合源码分析。环境准备与本地运行按照 README 的 Setup 章节本地运行这些示例需要以下步骤配置 Cloud Run 开发环境示例最终部署目标是 Cloud Run因此需要先完成 Cloud Run 的本地开发环境配置包括gcloudCLI、Docker、对目标 GCP 项目的权限等。安装 Python 工具链安装pip与virtualenv。若从零开始可参考 Google Cloud 官方的 Python 开发环境设置指南。创建并激活虚拟环境README 注明示例兼容 Python 2.7 与 3.4实际仓库依赖已更新至现代版本见下文依赖分析virtualenv env source env/bin/activate安装依赖pip install -r requirements.txt本地启动应用python main.py启动后Flask 开发服务器默认监听0.0.0.0:8080可通过PORT环境变量覆盖事件源或测试工具即可向/端点发送 POST 请求。依赖与运行配置分析从仓库实际的依赖清单可以确认各示例的运行环境eventarc/pubsub/requirements.txtFlask3.1.3、gunicorn23.0.0eventarc/audit-storage/requirements.txt在上述基础上增加cloudevents1.11.0CloudEvents 解析库eventarc/audit_iam/requirements.txt增加google-events0.14.0Google 事件 protobuf 类型、cloudevents1.11.0、googleapis-common-protos1.66.0eventarc/storage_handler/requirements.txt与 audit_iam 相同依赖google-events与cloudevents可见处理结构化事件审计日志、GCS 对象事件的示例需要额外的cloudevents与google-events依赖而 Pub/Sub 与 Generic 示例仅需 Flask 即可。所有示例在main.py中统一使用如下启动方式if __name__ __main__: app.run(debugTrue, host0.0.0.0, portint(os.environ.get(PORT, 8080)))示例一Pub/Sub 消息接收器核心实现eventarc/pubsub/main.py 定义了一个仅接受POST方法的 Flask 路由。Eventarc 投递的 Pub/Sub 消息以 JSON 结构到达关键字段为message.data——该字段是 Pub/Sub 消息 payload 的base64 编码字符串因此处理前必须先解码app.route(/, methods[POST]) def index(): data request.get_json() if not data: msg no Pub/Sub message received print(ferror: {msg}) return fBad Request: {msg}, 400 if not isinstance(data, dict) or message not in data: msg invalid Pub/Sub message format print(ferror: {msg}) return fBad Request: {msg}, 400 pubsub_message data[message] name World if isinstance(pubsub_message, dict) and data in pubsub_message: name base64.b64decode(pubsub_message[data]).decode(utf-8).strip() resp fHello, {name}! ID: {request.headers.get(ce-id)} print(resp) return (resp, 200)实现要点空 payload 防护请求体为空或非 JSON 时返回400避免服务异常结构校验要求顶层必须是 dict 且包含message键否则返回400base64 解码pubsub_message[data]需经base64.b64decode(...).decode(utf-8)还原为原始字符串再.strip()去除空白CloudEvent ID 回显通过request.headers.get(ce-id)读取 CloudEvent 标准头将事件 ID 拼入响应便于链路追踪。测试用例印证eventarc/pubsub/main_test.py 使用 Flask 内置的 test client 覆盖了五类场景与源码的校验逻辑一一对应test_empty_payload空 payload 返回400test_invalid_payload缺少message键返回400test_invalid_mimetype非 JSON 内容返回400test_minimally_valid_message仅{message: true}返回200并打印Hello, World! ID: {ce-id}test_populated_message构造 base64 编码的用户名验证解码结果正确出现在响应与日志中。测试通过binary_headers模拟 CloudEvent 二进制头ce-id、ce-type、ce-source、ce-specversion这与 Eventarc 实际投递时携带的 CloudEvent 头一致。示例二Audit Logs – Cloud Storage 审计日志接收器核心实现eventarc/audit-storage/main.py 演示了如何接收由 Cloud Storage 写操作产生的审计日志事件。与 Pub/Sub 示例不同这里引入cloudevents.http.from_http将原始 HTTP 请求转换为标准 CloudEvent 对象from cloudevents.http import from_http from flask import Flask, request app Flask(__name__) app.route(/, methods[POST]) def index(): # Create a CloudEvent object from the incoming request event from_http(request.headers, request.data) # Gets the GCS bucket name from the CloudEvent # Example: storage.googleapis.com/projects/_/buckets/my-bucket bucket event.get(subject) print(fDetected change in Cloud Storage bucket: {bucket}) return (fDetected change in Cloud Storage bucket: {bucket}, 200)实现要点CloudEvent 解析from_http(request.headers, request.data)自动从请求头与 body 重建 CloudEvent无需手写解析subject 语义CloudEvent 的subject属性携带事件主体信息。对于 Cloud Storage 审计日志其值为存储桶的完整资源名形如storage.googleapis.com/projects/_/buckets/my-bucket源码注释给出了这一格式示例业务响应将检测到的存储桶名拼入日志与 HTTP 响应返回200。测试用例印证eventarc/audit-storage/main_test.py 演示了 CloudEvents 库的反向用法——先用cloudevents.http.CloudEvent构造事件对象再用to_binary()序列化为二进制 HTTP 请求最后通过 test client 投递ce_attributes { id: str(uuid4), type: com.pytest.sample.event, source: my-test-source, specversion: 1.0, subject: test-bucket, } event CloudEvent(ce_attributes, dict()) headers, body to_binary(event) r client.post(/, headersheaders, databody) assert Detected change in Cloud Storage bucket: test-bucket in r.text测试断言subject属性被正确读取为存储桶名验证了event.get(subject)的取值链路。示例三Generic 通用事件接收器核心实现eventarc/generic/main.py 是一个回显型接收器用于调试与观察任意 Eventarc 事件的全貌。它打印事件请求头与请求体并将其原样返回app.route(/, methods[POST]) def index(): print(Event received!) print(HEADERS:) headers dict(request.headers) headers.pop(Authorization, None) # do not log authorization header if exists print(headers) print(BODY:) body dict(request.json) print(body) resp {headers: headers, body: body} return (resp, 200)实现要点全量回显将请求头与请求体同时写入日志并作为 JSON 响应返回便于开发者直观核对事件内容安全处理显式移除Authorization头避免将认证凭据打印进日志——这是事件接收器普遍需要遵守的安全实践适用场景当接入一种新的事件源、需要确认事件结构与字段名时Generic 接收器是最快的诊断工具。测试用例印证eventarc/generic/main_test.py 验证了回显逻辑的完整性断言日志中出现Event received!断言响应体中的头包含Ce-Specversion:1.0验证 CloudEvent 头被回显断言请求体内容{message: {data: Hello}}被完整打印。进阶变体审计日志的结构化处理仓库中还提供了两个超出 README 列表演示的进阶示例它们展示了同一主题下更深入的生产级用法。IAM 审计日志audit_iameventarc/audit_iam/main.py 将审计日志事件反序列化为强类型 protobuf 对象google.events.cloud.audit.LogEntryData实现检测服务账号密钥创建的业务逻辑event from_http(request.headers, request.get_data()) log_entry LogEntryData.from_json( json.dumps(event.get_data()), ignore_unknown_fieldsTrue ) if log_entry.proto_payload.service_name ! iam.googleapis.com: return (Received event was not from IAM., 400) if log_entry.proto_payload.status.code ! 0: return (Key creation failed, not reporting., 204) user log_entry.proto_payload.authentication_info.principal_email service_account log_entry.proto_payload.request[name] keypath log_entry.proto_payload.response[name] print(fNew Service Account Key created for {service_account} by {user}: {keypath})关键细节大小写转换LogEntryData.from_json内部将 JSON 风格的lowerCamelCase字段名转换为 protobuf 风格的snake_case未知字段容忍ignore_unknown_fieldsTrue用于跳过审计日志中的type等元数据字段多级过滤先校验服务名是否为iam.googleapis.com否则400再校验操作状态码code ! 0表示失败返回204不告警可直接操作的输出注释指出keypath可用于gcloud iam service-accounts keys disable ${keypath}等后续处置。Cloud Storage CloudEventstorage_handlereventarc/storage_handler/main.py 展示的是 Cloud Storage 对象级事件而非审计日志的处理方式使用google.events.cloud.storage.StorageObjectData强类型解析事件数据try: storage_obj StorageObjectData(event.data) gcs_object os.path.join(storage_obj.bucket, storage_obj.name) update_time storage_obj.updated return ( fCloud Storage object changed: {gcs_object} f updated at {update_time}, 200, ) except ValueError as e: return (fFailed to parse event data: {e}, 400)与审计日志示例相比这里直接从event.data构造StorageObjectData能够访问bucket、name、updated等对象属性解析失败ValueError时返回400。eventarc/storage_handler/main_test.py 同时验证了正常路径Cloud Storage object changed: test-bucket/my-file.txt与异常路径非法数据返回400。容器化部署Docker 与 Cloud Run所有示例都提供 Dockerfile用于将 Flask 应用打包为 Cloud Run 镜像。以 eventarc/audit-storage/Dockerfile 为例其构建步骤具有代表性基础镜像FROM python:3.14-slim使用官方精简 Python 镜像日志即时刷新ENV PYTHONUNBUFFERED True保证日志立即出现在 Cloud Run 日志中依赖分层缓存先单独COPY requirements.txt ./并RUN pip install -r requirements.txt避免每次代码变更都重新安装依赖工作目录ENV APP_HOME /app与WORKDIR $APP_HOME启动命令gunicorn 生产服务器CMD exec gunicorn --bind :$PORT --workers 1 --threads 8 --timeout 0 main:appDockerfile 注释明确解释了该命令的设计意图单 worker 8 线程若部署环境具有多 CPU 核心应将--workers调整为与核心数一致。--timeout 0用于禁用请求超时适配事件处理可能耗时较长的场景。$PORT由 Cloud Run 注入。eventarc/storage_handler/Procfile 则为非容器平台提供了等价的web: gunicorn --bind :$PORT --workers 1 --threads 8 --timeout 0 main:app启动声明。本地测试与验证每个示例均包含main_test.py测试框架统一使用 pytest见各目录的requirements-test.txt如 eventarc/pubsub/requirements-test.txt。运行方式pip install -r requirements.txt -r requirements-test.txt pytest测试的共性模式可以总结为三类模拟 CloudEvent 头通过字典构造ce-id、ce-type、ce-source、ce-specversion等标准头构造事件对象使用cloudevents.http.CloudEvent与cloudevents.conversion.to_binary将事件序列化为真实 HTTP 请求见 eventarc/audit-storage/main_test.py强类型数据构造对于依赖google-events的示例直接用StorageObjectData(bucket..., name...)、StorageObjectData.to_dict()构造合法负载见 eventarc/storage_handler/main_test.py。总结从 README 出发结合仓库源码可以梳理出 Eventarc Python 事件接收器的完整开发脉络三类核心模式Pub/Sub 消息的 base64 解码、审计日志的 CloudEvent 解析subject属性、Generic 的全量回显覆盖了事件接收的绝大多数场景两种解析路线轻量场景直接用request.get_json()手写解析生产级场景用cloudeventsgoogle-events强类型解析LogEntryData、StorageObjectData并借助ignore_unknown_fields、多级状态码过滤提升健壮性一套部署范式Flask 应用 gunicorn单 worker 多线程、--timeout 0 Dockerfile 分层缓存是 Cloud Run 事件接收服务的事实标准配置测试方法论以 pytest Flask test client 模拟 CloudEvent 二进制请求兼顾正常路径与400/204异常路径的断言。无论你是要接入第一条 Pub/Sub 事件还是需要处理结构复杂的 IAM 审计日志本目录的示例都可以直接作为脚手架结合 README 的 Setup 流程快速落地。赞分享示例工程【免费下载链接】python-docs-samplesCode samples used on cloud.google.com项目地址https://gitcode.com/GitHub_Trending/py/python-docs-samples点击查看免费下载相关推荐从零开始构建智能数字人助理Fay框架如何让AI真正为你工作从零开始构建智能数字人助理Fay框架如何让AI真正为你工作 想象一下每天早上醒来你的数字人助理已经为你规划好了一天的工作日程当你需要查询信息时它不仅能Nikto与Google Cloud Functions Eventarc集成事件驱动扫描新范式Nikto与Google Cloud Functions Eventarc集成事件驱动扫描新范式 痛点与解决方案 你是否还在为Web服务器安全检测的及时性和自rembg 人像分割实操指南从发丝边缘到批量证件照rembg 人像分割实操指南从发丝边缘到批量证件照 这篇文章讲的是 rembg 的人像背景去除这条线为什么默认模型抠人像总差点意思、birefnet por人工智能计算机视觉图像处理上一篇为什么选择nxdumptool3个超越传统备份方案的关键优势下一篇ChatGPT Retrieval Plugin 接入 Postgres pgvector环境配置、迁移与索引优化实战指南创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考 Eventarc Python 示例：基于 Cloud Run 的事件驱动接收器开发实战
示例工程【免费下载链接】python-docs-samplesCode samples used on cloud.google.com项目地址https://gitcode.com/GitHub_Trending/py/python-docs-samples点击查看免费下载导读本指南围绕 eventarc/README.md 展开介绍如何用 Flask 编写接收 CloudEvents 事件的服务端应用并部署到 Cloud Run用于响应 Pub/Sub 消息、Cloud Storage 审计日志等事件。读完本文你将掌握 Pub/Sub、审计日志Audit Logs与通用Generic三类事件接收器的核心实现、本地运行、自动化测试与容器化部署的完整实战方案。Eventarc 是什么Eventarc 是 Google Cloud 上统一的事件路由服务它将来自不同事件源如 Cloud Pub/Sub、Cloud Storage、IAM 审计日志等的事件统一封装为 CloudEvents 格式并投递到事件接收端点通常是一个 Cloud Run 服务。README 明确说明该目录包含用于 Eventarc 的 Python 示例这些示例以 Flask 为基础聚焦于接收事件并做出响应这一核心职责。示例全景README 将本目录的示例划分为三类对应 Eventarc 最常见的三种接入场景示例目录仓库根路径事件类型核心处理逻辑Pub/Subeventarc/pubsub/Cloud Pub/Sub 消息解码消息 payload返回问候语并携带事件 IDAudit Logs – Cloud Storageeventarc/audit-storage/Cloud Storage 写操作审计日志从 CloudEvent 中提取存储桶 subject 并打印Genericeventarc/generic/任意 HTTP 触发的 CloudEvent原样回显事件请求头与请求体此外仓库中还包含两个进阶变体示例audit_iamIAM 审计日志与storage_handlerCloud Storage CloudEvent 直连下文将一并结合源码分析。环境准备与本地运行按照 README 的 Setup 章节本地运行这些示例需要以下步骤配置 Cloud Run 开发环境示例最终部署目标是 Cloud Run因此需要先完成 Cloud Run 的本地开发环境配置包括gcloudCLI、Docker、对目标 GCP 项目的权限等。安装 Python 工具链安装pip与virtualenv。若从零开始可参考 Google Cloud 官方的 Python 开发环境设置指南。创建并激活虚拟环境README 注明示例兼容 Python 2.7 与 3.4实际仓库依赖已更新至现代版本见下文依赖分析virtualenv env source env/bin/activate安装依赖pip install -r requirements.txt本地启动应用python main.py启动后Flask 开发服务器默认监听0.0.0.0:8080可通过PORT环境变量覆盖事件源或测试工具即可向/端点发送 POST 请求。依赖与运行配置分析从仓库实际的依赖清单可以确认各示例的运行环境eventarc/pubsub/requirements.txtFlask3.1.3、gunicorn23.0.0eventarc/audit-storage/requirements.txt在上述基础上增加cloudevents1.11.0CloudEvents 解析库eventarc/audit_iam/requirements.txt增加google-events0.14.0Google 事件 protobuf 类型、cloudevents1.11.0、googleapis-common-protos1.66.0eventarc/storage_handler/requirements.txt与 audit_iam 相同依赖google-events与cloudevents可见处理结构化事件审计日志、GCS 对象事件的示例需要额外的cloudevents与google-events依赖而 Pub/Sub 与 Generic 示例仅需 Flask 即可。所有示例在main.py中统一使用如下启动方式if __name__ __main__: app.run(debugTrue, host0.0.0.0, portint(os.environ.get(PORT, 8080)))示例一Pub/Sub 消息接收器核心实现eventarc/pubsub/main.py 定义了一个仅接受POST方法的 Flask 路由。Eventarc 投递的 Pub/Sub 消息以 JSON 结构到达关键字段为message.data——该字段是 Pub/Sub 消息 payload 的base64 编码字符串因此处理前必须先解码app.route(/, methods[POST]) def index(): data request.get_json() if not data: msg no Pub/Sub message received print(ferror: {msg}) return fBad Request: {msg}, 400 if not isinstance(data, dict) or message not in data: msg invalid Pub/Sub message format print(ferror: {msg}) return fBad Request: {msg}, 400 pubsub_message data[message] name World if isinstance(pubsub_message, dict) and data in pubsub_message: name base64.b64decode(pubsub_message[data]).decode(utf-8).strip() resp fHello, {name}! ID: {request.headers.get(ce-id)} print(resp) return (resp, 200)实现要点空 payload 防护请求体为空或非 JSON 时返回400避免服务异常结构校验要求顶层必须是 dict 且包含message键否则返回400base64 解码pubsub_message[data]需经base64.b64decode(...).decode(utf-8)还原为原始字符串再.strip()去除空白CloudEvent ID 回显通过request.headers.get(ce-id)读取 CloudEvent 标准头将事件 ID 拼入响应便于链路追踪。测试用例印证eventarc/pubsub/main_test.py 使用 Flask 内置的 test client 覆盖了五类场景与源码的校验逻辑一一对应test_empty_payload空 payload 返回400test_invalid_payload缺少message键返回400test_invalid_mimetype非 JSON 内容返回400test_minimally_valid_message仅{message: true}返回200并打印Hello, World! ID: {ce-id}test_populated_message构造 base64 编码的用户名验证解码结果正确出现在响应与日志中。测试通过binary_headers模拟 CloudEvent 二进制头ce-id、ce-type、ce-source、ce-specversion这与 Eventarc 实际投递时携带的 CloudEvent 头一致。示例二Audit Logs – Cloud Storage 审计日志接收器核心实现eventarc/audit-storage/main.py 演示了如何接收由 Cloud Storage 写操作产生的审计日志事件。与 Pub/Sub 示例不同这里引入cloudevents.http.from_http将原始 HTTP 请求转换为标准 CloudEvent 对象from cloudevents.http import from_http from flask import Flask, request app Flask(__name__) app.route(/, methods[POST]) def index(): # Create a CloudEvent object from the incoming request event from_http(request.headers, request.data) # Gets the GCS bucket name from the CloudEvent # Example: storage.googleapis.com/projects/_/buckets/my-bucket bucket event.get(subject) print(fDetected change in Cloud Storage bucket: {bucket}) return (fDetected change in Cloud Storage bucket: {bucket}, 200)实现要点CloudEvent 解析from_http(request.headers, request.data)自动从请求头与 body 重建 CloudEvent无需手写解析subject 语义CloudEvent 的subject属性携带事件主体信息。对于 Cloud Storage 审计日志其值为存储桶的完整资源名形如storage.googleapis.com/projects/_/buckets/my-bucket源码注释给出了这一格式示例业务响应将检测到的存储桶名拼入日志与 HTTP 响应返回200。测试用例印证eventarc/audit-storage/main_test.py 演示了 CloudEvents 库的反向用法——先用cloudevents.http.CloudEvent构造事件对象再用to_binary()序列化为二进制 HTTP 请求最后通过 test client 投递ce_attributes { id: str(uuid4), type: com.pytest.sample.event, source: my-test-source, specversion: 1.0, subject: test-bucket, } event CloudEvent(ce_attributes, dict()) headers, body to_binary(event) r client.post(/, headersheaders, databody) assert Detected change in Cloud Storage bucket: test-bucket in r.text测试断言subject属性被正确读取为存储桶名验证了event.get(subject)的取值链路。示例三Generic 通用事件接收器核心实现eventarc/generic/main.py 是一个回显型接收器用于调试与观察任意 Eventarc 事件的全貌。它打印事件请求头与请求体并将其原样返回app.route(/, methods[POST]) def index(): print(Event received!) print(HEADERS:) headers dict(request.headers) headers.pop(Authorization, None) # do not log authorization header if exists print(headers) print(BODY:) body dict(request.json) print(body) resp {headers: headers, body: body} return (resp, 200)实现要点全量回显将请求头与请求体同时写入日志并作为 JSON 响应返回便于开发者直观核对事件内容安全处理显式移除Authorization头避免将认证凭据打印进日志——这是事件接收器普遍需要遵守的安全实践适用场景当接入一种新的事件源、需要确认事件结构与字段名时Generic 接收器是最快的诊断工具。测试用例印证eventarc/generic/main_test.py 验证了回显逻辑的完整性断言日志中出现Event received!断言响应体中的头包含Ce-Specversion:1.0验证 CloudEvent 头被回显断言请求体内容{message: {data: Hello}}被完整打印。进阶变体审计日志的结构化处理仓库中还提供了两个超出 README 列表演示的进阶示例它们展示了同一主题下更深入的生产级用法。IAM 审计日志audit_iameventarc/audit_iam/main.py 将审计日志事件反序列化为强类型 protobuf 对象google.events.cloud.audit.LogEntryData实现检测服务账号密钥创建的业务逻辑event from_http(request.headers, request.get_data()) log_entry LogEntryData.from_json( json.dumps(event.get_data()), ignore_unknown_fieldsTrue ) if log_entry.proto_payload.service_name ! iam.googleapis.com: return (Received event was not from IAM., 400) if log_entry.proto_payload.status.code ! 0: return (Key creation failed, not reporting., 204) user log_entry.proto_payload.authentication_info.principal_email service_account log_entry.proto_payload.request[name] keypath log_entry.proto_payload.response[name] print(fNew Service Account Key created for {service_account} by {user}: {keypath})关键细节大小写转换LogEntryData.from_json内部将 JSON 风格的lowerCamelCase字段名转换为 protobuf 风格的snake_case未知字段容忍ignore_unknown_fieldsTrue用于跳过审计日志中的type等元数据字段多级过滤先校验服务名是否为iam.googleapis.com否则400再校验操作状态码code ! 0表示失败返回204不告警可直接操作的输出注释指出keypath可用于gcloud iam service-accounts keys disable ${keypath}等后续处置。Cloud Storage CloudEventstorage_handlereventarc/storage_handler/main.py 展示的是 Cloud Storage 对象级事件而非审计日志的处理方式使用google.events.cloud.storage.StorageObjectData强类型解析事件数据try: storage_obj StorageObjectData(event.data) gcs_object os.path.join(storage_obj.bucket, storage_obj.name) update_time storage_obj.updated return ( fCloud Storage object changed: {gcs_object} f updated at {update_time}, 200, ) except ValueError as e: return (fFailed to parse event data: {e}, 400)与审计日志示例相比这里直接从event.data构造StorageObjectData能够访问bucket、name、updated等对象属性解析失败ValueError时返回400。eventarc/storage_handler/main_test.py 同时验证了正常路径Cloud Storage object changed: test-bucket/my-file.txt与异常路径非法数据返回400。容器化部署Docker 与 Cloud Run所有示例都提供 Dockerfile用于将 Flask 应用打包为 Cloud Run 镜像。以 eventarc/audit-storage/Dockerfile 为例其构建步骤具有代表性基础镜像FROM python:3.14-slim使用官方精简 Python 镜像日志即时刷新ENV PYTHONUNBUFFERED True保证日志立即出现在 Cloud Run 日志中依赖分层缓存先单独COPY requirements.txt ./并RUN pip install -r requirements.txt避免每次代码变更都重新安装依赖工作目录ENV APP_HOME /app与WORKDIR $APP_HOME启动命令gunicorn 生产服务器CMD exec gunicorn --bind :$PORT --workers 1 --threads 8 --timeout 0 main:appDockerfile 注释明确解释了该命令的设计意图单 worker 8 线程若部署环境具有多 CPU 核心应将--workers调整为与核心数一致。--timeout 0用于禁用请求超时适配事件处理可能耗时较长的场景。$PORT由 Cloud Run 注入。eventarc/storage_handler/Procfile 则为非容器平台提供了等价的web: gunicorn --bind :$PORT --workers 1 --threads 8 --timeout 0 main:app启动声明。本地测试与验证每个示例均包含main_test.py测试框架统一使用 pytest见各目录的requirements-test.txt如 eventarc/pubsub/requirements-test.txt。运行方式pip install -r requirements.txt -r requirements-test.txt pytest测试的共性模式可以总结为三类模拟 CloudEvent 头通过字典构造ce-id、ce-type、ce-source、ce-specversion等标准头构造事件对象使用cloudevents.http.CloudEvent与cloudevents.conversion.to_binary将事件序列化为真实 HTTP 请求见 eventarc/audit-storage/main_test.py强类型数据构造对于依赖google-events的示例直接用StorageObjectData(bucket..., name...)、StorageObjectData.to_dict()构造合法负载见 eventarc/storage_handler/main_test.py。总结从 README 出发结合仓库源码可以梳理出 Eventarc Python 事件接收器的完整开发脉络三类核心模式Pub/Sub 消息的 base64 解码、审计日志的 CloudEvent 解析subject属性、Generic 的全量回显覆盖了事件接收的绝大多数场景两种解析路线轻量场景直接用request.get_json()手写解析生产级场景用cloudeventsgoogle-events强类型解析LogEntryData、StorageObjectData并借助ignore_unknown_fields、多级状态码过滤提升健壮性一套部署范式Flask 应用 gunicorn单 worker 多线程、--timeout 0 Dockerfile 分层缓存是 Cloud Run 事件接收服务的事实标准配置测试方法论以 pytest Flask test client 模拟 CloudEvent 二进制请求兼顾正常路径与400/204异常路径的断言。无论你是要接入第一条 Pub/Sub 事件还是需要处理结构复杂的 IAM 审计日志本目录的示例都可以直接作为脚手架结合 README 的 Setup 流程快速落地。赞分享示例工程【免费下载链接】python-docs-samplesCode samples used on cloud.google.com项目地址https://gitcode.com/GitHub_Trending/py/python-docs-samples点击查看免费下载相关推荐从零开始构建智能数字人助理Fay框架如何让AI真正为你工作从零开始构建智能数字人助理Fay框架如何让AI真正为你工作 想象一下每天早上醒来你的数字人助理已经为你规划好了一天的工作日程当你需要查询信息时它不仅能Nikto与Google Cloud Functions Eventarc集成事件驱动扫描新范式Nikto与Google Cloud Functions Eventarc集成事件驱动扫描新范式 痛点与解决方案 你是否还在为Web服务器安全检测的及时性和自rembg 人像分割实操指南从发丝边缘到批量证件照rembg 人像分割实操指南从发丝边缘到批量证件照 这篇文章讲的是 rembg 的人像背景去除这条线为什么默认模型抠人像总差点意思、birefnet por人工智能计算机视觉图像处理上一篇为什么选择nxdumptool3个超越传统备份方案的关键优势下一篇ChatGPT Retrieval Plugin 接入 Postgres pgvector环境配置、迁移与索引优化实战指南创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考