ExternalTaskSensor: An option to use the data_interval variables rather than execution_date #26429

Activity

1. HeshamMeneisi

   added

   kind:featureFeature RequestsFeature Requests

   on Sep 16, 2022
2. uranusjr commented on Sep 16, 2022

   

   uranusjr

   on Sep 16, 2022

   Member

   More actions

   Makes sense. I believe there’s some discussion somewhere, but this write up is more complete. I think we should allow using either data_interval_start or data_interval_end; technically data_interval_start can be different from execution_date.

   To simplify implementation, perhaps it’s a good idea to split the logic into three sensors, one for each value? The three sensors would inherit from the same class that implements most of the logic, but inherited to applied on different values.
3. uranusjr

   assigned

   HeshamMeneisi

   on Sep 16, 2022
4. potiuk

   added

   good first issue

   on Sep 19, 2022
5. eladkal

   added

   area:core-operators

   on Sep 19, 2022
6. qcha41 commented on Oct 16, 2022

   

   qcha41

   on Oct 16, 2022

   Last edited by qcha41

   More actions

   Totally interested too! Any development on this already ?
7. potiuk

   assigned

   qcha41

   on Oct 24, 2022
8. potiuk commented on Oct 24, 2022

   

   potiuk

   on Oct 24, 2022

   Member

   More actions

   I assigned you too - so if you want ot help @qcha41 -> talk directly to @HeshamMeneisi
9. HeshamMeneisi commented on Jan 24, 2023

   

   HeshamMeneisi

   on Jan 24, 2023

   Last edited by HeshamMeneisi

   Author

   More actions

   I haven't got to do this specifically yet, but in light of the new Dataset logic, this seems like a more intuitive solution for the use-case above, what do you think? I can open a PR.

   class DatasetUpToDateSensor(BaseSensorOperator):
       """
       This sensor waits for the current data interval to be covered for a specific Dataset
   
       :param dataset: The target Dataset
       :param covered_to: The timestamp the dataset is desired to be covered to (updated at/after), defaults to data_interval_end (templated)
       """
   
       template_fields = ["covered_to"]
   
       def __init__(
           self, dataset: Dataset, covered_to: Optional[datetime.datetime] = None, **kwargs
       ) -> None:
           super().__init__(**kwargs)
           self.dataset = dataset
           self.covered_to = covered_to
   
       @provide_session
       def poke(self, context: Context, session=None) -> bool:
           dataset: DatasetModel = (
               session.query(DatasetModel)
               .filter(DatasetModel.uri == self.dataset.uri)
               .one_or_none()
           )
   
           if not dataset:
               raise DatasetNotFoundException(self.dataset)
   
           desired_covered_to = self.covered_to or context["data_interval_end"]
   
           return dataset.updated_at >= desired_covered_to

   In this case, covered_to can also be set to something like {{ logical_date + macros.timedelta(hours=1) }}, it's quite flexible.
10. potiuk commented on Feb 19, 2023

    

    potiuk

    on Feb 19, 2023

    Member

    More actions

    Opening a PR is a good idea.
11. eladkal

    unassigned

    HeshamMeneisi

    and

    qcha41

    on Aug 23, 2023
12. eladkal

    removed

    area:core-operators

    on May 20, 2025