uranusjr commented on code in PR #58543:
URL: https://github.com/apache/airflow/pull/58543#discussion_r3732275367
##########
airflow-core/src/airflow/timetables/assets.py:
##########
@@ -90,3 +100,77 @@ def generate_run_id(self, *, run_type: DagRunType,
**kwargs: typing.Any) -> str:
if run_type != DagRunType.ASSET_TRIGGERED:
return self.timetable.generate_run_id(run_type=run_type, **kwargs)
return super().generate_run_id(run_type=run_type, **kwargs)
+
+
+class AssetAndTimeSchedule(Timetable):
+ """
+ Time-based schedule that waits for required assets before creating a run.
+
+ This timetable composes a time-based timetable with an asset condition. It
+ schedules runs according to the provided ``timetable`` (e.g. cron), but a
run
+ is only created when all required assets are present. Unlike
+ :class:`AssetOrTimeSchedule`, this does not create asset-triggered runs.
+ """
+
+ asset_gated = True
+
+ def __init__(
+ self,
+ *,
+ timetable: Timetable,
+ assets: Collection[SerializedAsset] | SerializedAssetBase,
+ ) -> None:
+ self.timetable = timetable
+
+ if isinstance(assets, SerializedAssetBase | BaseAsset):
+ self.asset_condition = ensure_serialized_asset(assets)
Review Comment:
I believe SerializedAssetBase is not needed in this check?
--
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]
