o-nikolas commented on code in PR #66168:
URL: https://github.com/apache/airflow/pull/66168#discussion_r3174922200
##########
airflow-core/src/airflow/assets/manager.py:
##########
@@ -172,6 +172,68 @@ def _add_asset_alias_association(
if name not in already_related
)
+ @classmethod
+ def _filter_dags_by_team(
+ cls,
+ dags_to_queue: set[DagModel],
+ source_teams: AbstractSet[str],
+ asset_model: AssetModel,
+ source_is_api: bool,
+ *,
+ session: Session,
+ ) -> set[DagModel]:
+ """
+ Filter consuming DAGs based on team membership when multi_team is
enabled.
+
+ :param dags_to_queue: set of DagModel instances to potentially queue.
+ :param source_teams: set of team names the source belongs to. Empty
set means teamless.
+ :param asset_model: the AssetModel whose extra may contain
allow_producer_teams.
+ :param source_is_api: True if the event was triggered via the REST API
(not a DAG task).
+ :param session: SQLAlchemy session.
+ """
+ if not conf.getboolean("core", "multi_team"):
+ return dags_to_queue
+
+ if not dags_to_queue:
+ return dags_to_queue
+
+ from airflow.models.dag import DagModel
+
+ allow_producer_teams: list[str] =
asset_model.extra.get("allow_producer_teams", [])
+ is_teamless_source = len(source_teams) == 0
+
+ dag_ids = [dag.dag_id for dag in dags_to_queue]
+ dag_id_to_team = DagModel.get_dag_id_to_team_name_mapping(dag_ids,
session=session)
+
+ filtered = set()
+ for dag in dags_to_queue:
+ consumer_team = dag_id_to_team.get(dag.dag_id)
+
+ if consumer_team is None:
+ # Teamless consumer accepts events from any source
Review Comment:
Sure, let's chat, I may have had a different understanding before!
--
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]
