ramitkataria commented on code in PR #73569:
URL: https://github.com/apache/airflow/pull/73569#discussion_r4088232282
##########
airflow-core/src/airflow/dag_processing/dagbag.py:
##########
@@ -186,6 +180,53 @@ def _assign_default_team_pools(
task.pool = Pool.get_default_team_pool_name(dag_team_name)
+def _validate_plugin_scheduling_classes(dag: DAG, bundle_name: str | None =
None) -> None:
+ """
+ Reject a Dag that uses scheduling classes belonging to another team's
plugin.
+
+ Timetables and priority weight strategies are imported and instantiated by
the Dag
+ author, so unlike executors or pools there is no team-aware lookup in the
way. This
+ check is the only thing keeping a team-scoped plugin's scheduling classes
from being
+ used by Dags outside that team.
+
+ Raising here surfaces as an import error for the Dag, leaving the rest of
the bundle
+ to parse normally.
+ """
+ if not conf.getboolean("core", "multi_team"):
+ return
+
+ from airflow.plugins_manager import owning_teams_of_scheduling_class
+
+ dag_team_name = _bundle_team_name(bundle_name)
+
+ candidates: list[tuple[str, Any]] = [("its timetable", dag.timetable)]
Review Comment:
Does this/should we account for inner timetables as well?
##########
airflow-core/src/airflow/dag_processing/dagbag.py:
##########
@@ -186,6 +180,53 @@ def _assign_default_team_pools(
task.pool = Pool.get_default_team_pool_name(dag_team_name)
+def _validate_plugin_scheduling_classes(dag: DAG, bundle_name: str | None =
None) -> None:
+ """
+ Reject a Dag that uses scheduling classes belonging to another team's
plugin.
+
+ Timetables and priority weight strategies are imported and instantiated by
the Dag
+ author, so unlike executors or pools there is no team-aware lookup in the
way. This
+ check is the only thing keeping a team-scoped plugin's scheduling classes
from being
+ used by Dags outside that team.
+
+ Raising here surfaces as an import error for the Dag, leaving the rest of
the bundle
+ to parse normally.
+ """
+ if not conf.getboolean("core", "multi_team"):
+ return
+
+ from airflow.plugins_manager import owning_teams_of_scheduling_class
+
+ dag_team_name = _bundle_team_name(bundle_name)
+
+ candidates: list[tuple[str, Any]] = [("its timetable", dag.timetable)]
+ for task in dag.tasks:
+ # A string weight_rule names a built-in strategy, so only instances
can come
Review Comment:
Hmm, what about something like this?
https://github.com/apache/airflow/blob/63b615b252589ef49b8504f2dbe2942b543ec896/airflow-core/src/airflow/example_dags/example_custom_weight.py#L52
--
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]
