uranusjr commented on code in PR #68778:
URL: https://github.com/apache/airflow/pull/68778#discussion_r3449016861
##########
airflow-core/src/airflow/jobs/scheduler_job_runner.py:
##########
@@ -2261,23 +2261,35 @@ def _create_dagruns_for_partitioned_asset_dags(self,
session: Session) -> set[st
source_key_by_asset = source_key_by_asset_per_apdr[apdr.id]
timetable = dag.timetable
- statuses: dict[SerializedAssetUniqueKey, bool] = {}
- for asset_id, (name, uri) in asset_info_per_apdr[apdr.id].items():
- key = SerializedAssetUniqueKey(name=name, uri=uri)
- if timetable.partitioned:
- statuses[key] = self._resolve_asset_partition_status(
- session=session,
- asset_id=asset_id,
- name=name,
- uri=uri,
- apdr=apdr,
- timetable=timetable,
- actual_by_asset=source_key_by_asset,
- )
- else:
- statuses[key] = True
- if not evaluator.run(timetable.asset_condition, statuses=statuses):
- continue
+ contributing_assets = asset_info_per_apdr[apdr.id]
+ if apdr.is_refresh:
Review Comment:
Instead of storing a simple boolean, it may be a good idea to store the
retun policy directly on the model (as a string), and resolve the stored value
to a full RerunPolicy object. That way, this block can be extracted into a
function on RerunPolicy. Making the logic live on RerunPolicy is nicer because
it is easier to keep track of possible enum values on the same class, and if we
potentially add new values to the enum, it is much easier to remember to add a
branch if the logic is close.
--
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]
