shahar1 commented on code in PR #73103:
URL: https://github.com/apache/airflow/pull/73103#discussion_r4019882731
##########
providers/google/src/airflow/providers/google/cloud/sensors/cloud_composer.py:
##########
@@ -426,49 +461,26 @@ def __init__(
"use `composer_external_task_id` or
`composer_external_task_ids` or `composer_external_task_group_id`."
)
- # check the requested states are all valid states for the target type,
be it dag or task
+ total_states = set(self.allowed_states + self.skipped_states +
self.failed_states)
if composer_external_task_ids or composer_external_task_group_id:
if not total_states <= set(State.task_states):
raise ValueError(
"Valid values for `allowed_states`, `skipped_states` and
`failed_states` "
- "when `composer_external_task_id` or
`composer_external_task_ids` or `composer_external_task_group_id` "
- f"is not `None`: {State.task_states}"
+ "when `composer_external_task_id` or
`composer_external_task_ids` or "
+ f"`composer_external_task_group_id` is not `None`:
{State.task_states}"
)
elif not total_states <= set(State.dag_states):
raise ValueError(
"Valid values for `allowed_states`, `skipped_states` and
`failed_states` "
- f"when `composer_external_task_id` and
`composer_external_task_group_id` is `None`: {State.dag_states}"
+ f"when `composer_external_task_id` and
`composer_external_task_group_id` is `None`: "
+ f"{State.dag_states}"
)
- self.execution_range = execution_range
- self.composer_external_dag_id = composer_external_dag_id
- self.composer_external_task_id = composer_external_task_id
self.composer_external_task_ids = composer_external_task_ids
- self.composer_external_task_group_id = composer_external_task_group_id
- self.gcp_conn_id = gcp_conn_id
- self.impersonation_chain = impersonation_chain
- self.deferrable = deferrable
- self.poll_interval = poll_interval
-
- def _get_logical_dates(self, context) -> tuple[datetime, datetime]:
- logical_date = context.get("logical_date", None)
- if logical_date is None:
- raise RuntimeError(
- "logical_date is None. Please make sure the sensor is not used
in an asset-triggered Dag. "
- "CloudComposerDAGRunSensor was designed to be used in
time-based scheduled Dags only, "
- "and asset-triggered Dags do not have logical_date. "
- )
- if isinstance(self.execution_range, timedelta):
- if self.execution_range < timedelta(0):
- return logical_date, logical_date - self.execution_range
- return logical_date - self.execution_range, logical_date
- if isinstance(self.execution_range, list) and
len(self.execution_range) > 0:
- return self.execution_range[0], self.execution_range[1] if len(
- self.execution_range
- ) > 1 else logical_date
- return logical_date - timedelta(1), logical_date
+ self._external_task_ids_normalized = True
def poke(self, context: Context) -> bool:
+ self._normalize_external_task_ids()
Review Comment:
Deferrable tasks resume on a fresh instance directly through
`execute_complete()`, bypassing normalization in `execute()` and `poke()`. With
only `composer_external_task_id` set, a failed event therefore raises
`ExternalDagFailedError` instead of `ExternalTaskFailedError`.
Please call `_normalize_external_task_ids()` at the start of
`execute_complete()` and add a regression test that calls it directly on a
fresh deferrable instance with `composer_external_task_id="task_a"`, asserting
`ExternalTaskFailedError` for `event={"status": "failed"}`.
##########
providers/google/src/airflow/providers/google/cloud/sensors/cloud_composer.py:
##########
@@ -426,49 +461,26 @@ def __init__(
"use `composer_external_task_id` or
`composer_external_task_ids` or `composer_external_task_group_id`."
)
- # check the requested states are all valid states for the target type,
be it dag or task
+ total_states = set(self.allowed_states + self.skipped_states +
self.failed_states)
Review Comment:
This is now calcualted twice
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
