o-nikolas commented on code in PR #66168:
URL: https://github.com/apache/airflow/pull/66168#discussion_r3171301965
##########
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:
Is this right? If a Dag is not assigned to a team, then it can read from any
event source? Wouldn't that allow someone to get around a team restriction by
just making their dag global?
##########
airflow-core/src/airflow/assets/manager.py:
##########
@@ -182,13 +244,28 @@ def register_asset_change(
source_alias_names: Collection[str] = (),
session: Session,
partition_key: str | None = None,
+ source_is_api: bool = False,
+ api_user_teams: set[str] | None = None,
**kwargs,
) -> AssetEvent | None:
"""
Register asset related changes.
For local assets, look them up, record the asset event, queue dagruns,
and broadcast
- the asset event
+ the asset event.
+
+ When multi_team mode is enabled, team-based filtering is applied to
determine which
+ consumer DAGs should be queued:
+ - For DAG-produced events (task_instance is set), source teams are
resolved automatically
+ from the producing DAG's bundle.
+ - For API-produced events (source_is_api=True), the caller must pass
api_user_teams
+ obtained from the auth manager.
Review Comment:
I'm confused, so is it passed by the caller or fetched from the auth
manager? This says both?
--
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]
