ashb commented on a change in pull request #15125:
URL: https://github.com/apache/airflow/pull/15125#discussion_r605492209
##########
File path: airflow/utils/decorators.py
##########
@@ -97,6 +99,44 @@ def wrapper(*args: Any, **kwargs: Any) -> Any:
return cast(T, wrapper)
+def latest_only(f):
+ """
+ Decorator for adding skip-if-not-latest behavior to any operator.
+
+ Can be disabled if the class has an attribute ``latest_only`` with value
``False``
+ """
+
+ def skip_if_not_latest(context):
+ if not context: # assume run interactively
+ return
+ dag_run = context.get('dag_run')
+ if dag_run and dag_run.external_trigger:
+ print("Externally triggered DAG_Run: allowing execution to
proceed.")
+ return
+
+ dag = context['dag']
+ now = pendulum.now('UTC')
+ left_window = dag.following_schedule(context['execution_date'])
+ right_window = dag.following_schedule(left_window)
+ print(
+ f"Checking latest only:\n"
+ f"\tleft_window: {left_window}\n"
+ f"\tright_window: {right_window}\n"
+ f"\tnow: {now}\n",
+ )
+
+ if not left_window < now <= right_window:
+ raise AirflowSkipException('Not latest execution; skipping...')
+
+ @wraps(f)
+ def wrap(self, context):
+ if not (hasattr(self, 'latest_only') and self.latest_only is False):
Review comment:
Why do we need the attribute at all here? Surely if you don't want it,
then don't apply the decorator?
--
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.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
