Source code for tests.system.providers.google.cloud.cloud_functions.example_functions
## Licensed to the Apache Software Foundation (ASF) under one# or more contributor license agreements. See the NOTICE file# distributed with this work for additional information# regarding copyright ownership. The ASF licenses this file# to you under the Apache License, Version 2.0 (the# "License"); you may not use this file except in compliance# with the License. You may obtain a copy of the License at## http://www.apache.org/licenses/LICENSE-2.0## Unless required by applicable law or agreed to in writing,# software distributed under the License is distributed on an# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY# KIND, either express or implied. See the License for the# specific language governing permissions and limitations# under the License."""Example Airflow DAG that displays interactions with Google Cloud Functions.It creates a function and then deletes it.This DAG relies on the following OS environment variableshttps://airflow.apache.org/concepts.html#variables* PROJECT_ID - Google Cloud Project to use for the Cloud Function.* LOCATION - Google Cloud Functions region where the function should be created.* ENTRYPOINT - Name of the executable function in the source code.* and one of the below: * SOURCE_ARCHIVE_URL - Path to the zipped source in Google Cloud Storage * SOURCE_UPLOAD_URL - Generated upload URL for the zipped source and ZIP_PATH - Local path to the zipped source archive * SOURCE_REPOSITORY - The URL pointing to the hosted repository where the function is defined in a supported Cloud Source Repository URL format https://cloud.google.com/functions/docs/reference/rest/v1/projects.locations.functions#SourceRepository"""from__future__importannotationsimportosfromdatetimeimportdatetimefromtypingimportAnyfromairflowimportmodelsfromairflow.models.baseoperatorimportchainfromairflow.providers.google.cloud.operators.functionsimport(CloudFunctionDeleteFunctionOperator,CloudFunctionDeployFunctionOperator,CloudFunctionInvokeFunctionOperator,)
# [END howto_operator_gcf_default_args]# [START howto_operator_gcf_deploy_variants]ifSOURCE_ARCHIVE_URL:body["sourceArchiveUrl"]=SOURCE_ARCHIVE_URLelifSOURCE_REPOSITORY:body["sourceRepository"]={"url":SOURCE_REPOSITORY}elifZIP_PATH:body["sourceUploadUrl"]=""default_args["zip_path"]=ZIP_PATHelifSOURCE_UPLOAD_URL:body["sourceUploadUrl"]=SOURCE_UPLOAD_URLelse:raiseException("Please provide one of the source_code parameters")# [END howto_operator_gcf_deploy_variants]withmodels.DAG(DAG_ID,default_args=default_args,start_date=datetime(2021,1,1),catchup=False,tags=["example"],)asdag:# [START howto_operator_gcf_deploy]
)# [END howto_operator_gcf_deploy]# [START howto_operator_gcf_deploy_no_project_id]deploy2_task=CloudFunctionDeployFunctionOperator(task_id="gcf_deploy2_task",location=LOCATION,body=body,validate_body=VALIDATE_BODY)# [END howto_operator_gcf_deploy_no_project_id]# [START howto_operator_gcf_invoke_function]invoke_task=CloudFunctionInvokeFunctionOperator(task_id="invoke_task",project_id=PROJECT_ID,location=LOCATION,input_data={},function_id=SHORT_FUNCTION_NAME,)# [END howto_operator_gcf_invoke_function]# [START howto_operator_gcf_delete]delete_task=CloudFunctionDeleteFunctionOperator(task_id="gcf_delete_task",name=FUNCTION_NAME)# [END howto_operator_gcf_delete]chain(deploy_task,deploy2_task,invoke_task,delete_task,)fromtests.system.utils.watcherimportwatcher# This test needs watcher in order to properly mark success/failure# when "tearDown" task with trigger rule is part of the DAGlist(dag.tasks)>>watcher()fromtests.system.utilsimportget_test_run# noqa: E402# Needed to run the example DAG with pytest (see: tests/system/README.md#run_via_pytest)