nailo2c commented on code in PR #58543:
URL: https://github.com/apache/airflow/pull/58543#discussion_r3613349260
##########
airflow-core/src/airflow/timetables/assets.py:
##########
@@ -90,3 +93,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)
+ else:
+ self.asset_condition =
SerializedAssetAll([ensure_serialized_asset(a) for a in assets])
+
+ self.description = f"Triggered by assets and {timetable.description}"
+ self.periodic = timetable.periodic
+ self.can_be_scheduled = timetable.can_be_scheduled
+ self.active_runs_limit = timetable.active_runs_limit
+
+ @classmethod
+ def deserialize(cls, data: dict[str, typing.Any]) -> Timetable:
+ from airflow.serialization.decoders import decode_asset_like,
decode_timetable
+
+ return cls(
+ assets=decode_asset_like(data["asset_condition"]),
+ timetable=decode_timetable(data["timetable"]),
+ )
+
+ def serialize(self) -> dict[str, typing.Any]:
+ from airflow.serialization.encoders import encode_asset_like,
encode_timetable
+
+ return {
+ "asset_condition": encode_asset_like(self.asset_condition),
+ "timetable": encode_timetable(self.timetable),
+ }
+
+ def validate(self) -> None:
+ if self.timetable.asset_triggered or self.timetable.asset_gated:
+ raise AirflowTimetableInvalid("cannot nest asset-aware timetables")
+ if not isinstance(self.asset_condition, SerializedAssetBase):
+ raise AirflowTimetableInvalid("all elements in 'assets' must be
assets")
Review Comment:
Good idea, let me do it later :)
--
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]
