Google Cloud Functions Operators¶

前提任务¶

要使用这些 operators，您必须执行以下几项操作

使用 Cloud Console 选择或创建一个 Cloud Platform 项目。

为您的项目启用结算功能，详情请参阅 Google Cloud 文档。

启用 API，详情请参阅 Cloud Console 文档。
通过 pip 安装 API 库。
pip install 'apache-airflow[google]'
详细信息请参阅安装。
设置 Google Cloud 连接.

CloudFunctionDeleteFunctionOperator¶

使用此 operator 从 Google Cloud Functions 中删除函数。

有关参数定义，请参阅 CloudFunctionDeleteFunctionOperator。

使用此 operator¶

tests/system/google/cloud/cloud_functions/example_functions.py

delete_function = CloudFunctionDeleteFunctionOperator(task_id="delete_function", name=FUNCTION_NAME)

模板化¶

template_fields: Sequence[str] = (
    "name",
    "gcp_conn_id",
    "api_version",
    "impersonation_chain",
)

CloudFunctionDeployFunctionOperator¶

使用此 operator 将函数部署到 Google Cloud Functions。如果同名函数已存在，则会进行更新。

有关参数定义，请参阅 CloudFunctionDeployFunctionOperator。

参数¶

创建 DAG 时，可以使用 default_args 字典传递与其他任务通用的参数

tests/system/google/cloud/cloud_functions/example_functions.py

default_args: dict[str, Any] = {"retries": 3}

请注意，上述示例中的 body 和 default_args 都不完整。根据设置的变量，传递源代码相关字段的方式可能有所不同。目前，您可以传递 sourceArchiveUrl、sourceRepository 或 sourceUploadUrl，如 Cloud Functions API 规范中所述。

此外，default_args 或直接的 operator 参数可能包含 zip_path 参数，以便在部署前执行上传源代码的额外步骤。在这种情况下，您还需要在 body 中提供一个空的 sourceUploadUrl 参数。

使用此 operator¶

根据参数的组合，函数的源代码可以从不同的来源获取

tests/system/google/cloud/cloud_functions/example_functions.py

body = {"name": FUNCTION_NAME, "entryPoint": ENTRYPOINT, "runtime": RUNTIME, "httpsTrigger": {}}

tests/system/google/cloud/cloud_functions/example_functions.py

if SOURCE_ARCHIVE_URL:
    body["sourceArchiveUrl"] = SOURCE_ARCHIVE_URL
elif SOURCE_REPOSITORY:
    body["sourceRepository"] = {"url": SOURCE_REPOSITORY}
elif ZIP_PATH:
    body["sourceUploadUrl"] = ""
    default_args["zip_path"] = ZIP_PATH
elif SOURCE_UPLOAD_URL:
    body["sourceUploadUrl"] = SOURCE_UPLOAD_URL
else:
    raise Exception("Please provide one of the source_code parameters")

创建 operator 的代码

tests/system/google/cloud/cloud_functions/example_functions.py

deploy_function = CloudFunctionDeployFunctionOperator(
    task_id="deploy_function",
    project_id=PROJECT_ID,
    location=LOCATION,
    body=body,
    validate_body=VALIDATE_BODY,
)

您也可以在不指定项目 ID 的情况下创建 operator - 项目 ID 将从使用的 Google Cloud 连接中检索

tests/system/google/cloud/cloud_functions/example_functions.py

deploy_function_no_project = CloudFunctionDeployFunctionOperator(
    task_id="deploy_function_no_project", location=LOCATION, body=body, validate_body=VALIDATE_BODY
)

模板化¶

template_fields: Sequence[str] = (
    "body",
    "project_id",
    "location",
    "gcp_conn_id",
    "api_version",
    "impersonation_chain",
)

故障排除¶

如果在部署期间看到类似如下错误

“HttpError 403: Missing necessary permission iam.serviceAccounts.actAs for on resource project-name@appspot.gserviceaccount.com. Please grant the roles/iam.serviceAccountUser role.”

这意味着您的服务帐号没有正确的 Cloud IAM 权限。

将 Cloud Functions Developer 角色分配给您的服务帐号。
授予用户 Cloud IAM Service Account User 角色给 Cloud Functions 运行时服务帐号。

使用 gcloud 分配 Cloud IAM 权限的典型方法如下所示。只需将 PROJECT_ID 替换为您的 Google Cloud 项目 ID，将 SERVICE_ACCOUNT_EMAIL 替换为您的服务帐号邮箱 ID。

gcloud iam service-accounts add-iam-policy-binding \
  PROJECT_ID@appspot.gserviceaccount.com \
  --member="serviceAccount:[SERVICE_ACCOUNT_EMAIL]" \
  --role="roles/iam.serviceAccountUser"

您也可以通过 Google Cloud Console 执行此操作。

详情请参阅向运行时服务添加 IAM 服务代理用户角色。

如果您的函数源代码位于 Google Source Repository 中，请确保您的服务帐号具有 Source Repository Viewer 角色，以便在需要时可以下载源代码。

参考¶

有关更多信息，请参阅

Google Cloud Functions Operators¶

前提任务¶

CloudFunctionDeleteFunctionOperator¶

使用此 operator¶

模板化¶

更多信息¶

CloudFunctionDeployFunctionOperator¶

参数¶

使用此 operator¶

模板化¶

故障排除¶

更多信息¶

参考¶