airflow.providers.google.cloud.sensors.cloud_composer¶

This module contains a Cloud Composer sensor.

Classes¶

`CloudComposerDAGRunSensor`	Check if a DAG run has completed.
`CloudComposerExternalTaskSensor`	Waits for a different DAG, task group, or task to complete for a specific composer environment.

Module Contents¶

class airflow.providers.google.cloud.sensors.cloud_composer.CloudComposerDAGRunSensor(*, project_id, region, environment_id, composer_dag_id, allowed_states=None, execution_range=None, composer_dag_run_id=None, gcp_conn_id='google_cloud_default', impersonation_chain=None, deferrable=conf.getboolean('operators', 'default_deferrable', fallback=False), poll_interval=10, use_rest_api=False, **kwargs)[source]¶

Bases: airflow.providers.common.compat.sdk.BaseSensorOperator

Check if a DAG run has completed.

Parameters:

project_id (str) – Required. The ID of the Google Cloud project that the service belongs to.
region (str) – Required. The ID of the Google Cloud region that the service belongs to.
environment_id (str) – The name of the Composer environment.
composer_dag_id (str) – The ID of executable DAG.
allowed_states (collections.abc.Iterable[str] | None) – Iterable of allowed states, default is ['success'].
execution_range (datetime.timedelta | list[datetime.datetime] | None) – execution DAGs time range. Sensor checks DAGs states only for DAGs which were started in this time range. For yesterday, use [positive!] datetime.timedelta(days=1). For future, use [negative!] datetime.timedelta(days=-1). For specific time, use list of datetimes [datetime(2024,3,22,11,0,0), datetime(2024,3,22,12,0,0)]. Or [datetime(2024,3,22,0,0,0)] in this case sensor will check for states from specific time in the past till current time execution. Default value datetime.timedelta(days=1).
composer_dag_run_id (str | None) – The Run ID of executable task. The ‘execution_range’ param is ignored, if both specified.
gcp_conn_id (str) – The connection ID to use when fetching connection info.
impersonation_chain (str | collections.abc.Sequence[str] | None) – Optional service account to impersonate using short-term credentials, or chained list of accounts required to get the access_token of the last account in the list, which will be impersonated in the request. If set as a string, the account must grant the originating account the Service Account Token Creator IAM role. If set as a sequence, the identities from the list must grant Service Account Token Creator IAM role to the directly preceding identity, with first account from the list granting this role to the originating account (templated).
poll_interval (int) – Optional: Control the rate of the poll for the result of deferrable run.
deferrable (bool) – Run sensor in deferrable mode.

template_fields = ('project_id', 'region', 'environment_id', 'composer_dag_id', 'impersonation_chain')[source]¶

project_id[source]¶

region[source]¶

environment_id[source]¶

composer_dag_id[source]¶

allowed_states[source]¶

execution_range = None[source]¶

composer_dag_run_id = None[source]¶

gcp_conn_id = 'google_cloud_default'[source]¶

impersonation_chain = None[source]¶

deferrable = True[source]¶

poll_interval = 10[source]¶

use_rest_api = False[source]¶

poke(context)[source]¶

Override when deriving this class.

execute(context)[source]¶

Derive when creating an operator.

The main method to execute the task. Context is the same dictionary used as when rendering jinja templates.

Refer to get_template_context for more context.

execute_complete(context, event)[source]¶

property hook: airflow.providers.google.cloud.hooks.cloud_composer.CloudComposerHook[source]¶

class airflow.providers.google.cloud.sensors.cloud_composer.CloudComposerExternalTaskSensor(*, project_id, region, environment_id, composer_external_dag_id, composer_external_task_id=None, composer_external_task_ids=None, composer_external_task_group_id=None, allowed_states=None, skipped_states=None, failed_states=None, execution_range=None, gcp_conn_id='google_cloud_default', impersonation_chain=None, deferrable=conf.getboolean('operators', 'default_deferrable', fallback=False), poll_interval=10, **kwargs)[source]¶

Bases: airflow.providers.common.compat.sdk.BaseSensorOperator

Waits for a different DAG, task group, or task to complete for a specific composer environment.

If both composer_external_task_group_id and composer_external_task_id are None (default), the sensor waits for the DAG. Values for composer_external_task_group_id and composer_external_task_id can’t be set at the same time.

By default, the CloudComposerExternalTaskSensor will wait for the external task to succeed, at which point it will also succeed. However, by default it will not fail if the external task fails, but will continue to check the status until the sensor times out (thus giving you time to retry the external task without also having to clear the sensor).

By default, the CloudComposerExternalTaskSensor will not skip if the external task skips. To change this, simply set skipped_states=[TaskInstanceState.SKIPPED]. Note that if you are monitoring multiple tasks, and one enters error state and the other enters a skipped state, then the external task will react to whichever one it sees first. If both happen together, then the failed state takes priority.

It is possible to alter the default behavior by setting states which cause the sensor to fail, e.g. by setting allowed_states=[DagRunState.FAILED] and failed_states=[DagRunState.SUCCESS] you will flip the behaviour to get a sensor which goes green when the external task fails and immediately goes red if the external task succeeds!

Note that soft_fail is respected when examining the failed_states. Thus if the external task enters a failed state and soft_fail == True the sensor will _skip_ rather than fail. As a result, setting soft_fail=True and failed_states=[DagRunState.SKIPPED] will result in the sensor skipping if the external task skips. However, this is a contrived example—consider using skipped_states if you would like this behaviour. Using skipped_states allows the sensor to skip if the target fails, but still enter failed state on timeout. Using soft_fail == True as above will cause the sensor to skip if the target fails, but also if it times out.