ds = get_next_ds()
        if ds is None:
            return ds
        return ds.replace("-", "")
    @cache
    def get_prev_execution_date():
        
        
        
        
        if dag_run.external_trigger:
            return logical_date
        with warnings.catch_warnings():
            warnings.simplefilter("ignore", RemovedInAirflow3Warning)
            return dag.previous_schedule(logical_date)
    @cache
    def get_prev_ds() -> str | None:
        execution_date = get_prev_execution_date()
        if execution_date is None:
            return None
        return execution_date.strftime("%Y-%m-%d")
    def get_prev_ds_nodash() -> str | None:
        prev_ds = get_prev_ds()
        if prev_ds is None:
            return None
        return prev_ds.replace("-", "")
    def get_triggering_events() -> dict[str, list[DatasetEvent]]: