This is an automated email from the ASF dual-hosted git repository.
o-nikolas pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/airflow.git
The following commit(s) were added to refs/heads/main by this push:
new 8c3a1169709 Add team_name attribute to plugins for multi-team (#69502)
8c3a1169709 is described below
commit 8c3a11697099f46dee05826e12084d80fe2769f0
Author: Niko Oliveira <[email protected]>
AuthorDate: Wed Jul 15 12:24:14 2026 -0700
Add team_name attribute to plugins for multi-team (#69502)
* Add team_name attribute to plugins for multi-team
First step for Plugin team awareness. This adds a team_name attribute to
AirflowPlugin so a plugin can be scoped to a team, exposes it through the
plugins API, and filters the /api/v2/plugins listing so users only see
global plugins and plugins for teams they are authorized for when
multi-team mode is enabled. The API server also validates at startup that
every team-scoped plugin references an existing team.
* Add team_name to expected plugin dict in CLI tests
* Harden assertions on DB test
---
airflow-core/src/airflow/api_fastapi/app.py | 3 +
.../api_fastapi/core_api/datamodels/plugins.py | 1 +
.../core_api/openapi/v2-rest-api-generated.yaml | 5 ++
.../api_fastapi/core_api/routes/public/plugins.py | 16 ++++-
airflow-core/src/airflow/plugins_manager.py | 39 +++++++++++-
.../airflow/ui/openapi-gen/requests/schemas.gen.ts | 11 ++++
.../airflow/ui/openapi-gen/requests/types.gen.ts | 1 +
.../core_api/routes/public/test_plugins.py | 73 ++++++++++++++++++++++
.../unit/cli/commands/test_plugins_command.py | 1 +
.../tests/unit/plugins/test_plugins_manager.py | 68 ++++++++++++++++++++
.../src/airflowctl/api/datamodels/generated.py | 1 +
.../plugins_manager/plugins_manager.py | 8 +++
.../tests/plugins_manager/test_plugins_manager.py | 16 +++++
13 files changed, 241 insertions(+), 2 deletions(-)
diff --git a/airflow-core/src/airflow/api_fastapi/app.py
b/airflow-core/src/airflow/api_fastapi/app.py
index a4dbc3c9b72..6e6bca84389 100644
--- a/airflow-core/src/airflow/api_fastapi/app.py
+++ b/airflow-core/src/airflow/api_fastapi/app.py
@@ -132,8 +132,11 @@ def create_app(apps: str = "all") -> FastAPI:
app.mount("/execution", task_exec_api_app)
if "all" in apps_list or "core" in apps_list:
+ from airflow import plugins_manager
+
app.state.dag_bag = dag_bag
init_plugins(app)
+ plugins_manager.validate_plugin_teams()
init_auth_manager(app)
init_flask_plugins(app)
init_views(app) # Core views need to be the last routes added - it
has a catch all route
diff --git
a/airflow-core/src/airflow/api_fastapi/core_api/datamodels/plugins.py
b/airflow-core/src/airflow/api_fastapi/core_api/datamodels/plugins.py
index 2bddb29ac96..ed2e9d6a86f 100644
--- a/airflow-core/src/airflow/api_fastapi/core_api/datamodels/plugins.py
+++ b/airflow-core/src/airflow/api_fastapi/core_api/datamodels/plugins.py
@@ -107,6 +107,7 @@ class PluginResponse(BaseModel):
"""Plugin serializer."""
name: str
+ team_name: str | None = None
macros: list[str]
flask_blueprints: list[str]
fastapi_apps: list[FastAPIAppResponse]
diff --git
a/airflow-core/src/airflow/api_fastapi/core_api/openapi/v2-rest-api-generated.yaml
b/airflow-core/src/airflow/api_fastapi/core_api/openapi/v2-rest-api-generated.yaml
index 005da0c6f4b..af18cab351e 100644
---
a/airflow-core/src/airflow/api_fastapi/core_api/openapi/v2-rest-api-generated.yaml
+++
b/airflow-core/src/airflow/api_fastapi/core_api/openapi/v2-rest-api-generated.yaml
@@ -15381,6 +15381,11 @@ components:
name:
type: string
title: Name
+ team_name:
+ anyOf:
+ - type: string
+ - type: 'null'
+ title: Team Name
macros:
items:
type: string
diff --git
a/airflow-core/src/airflow/api_fastapi/core_api/routes/public/plugins.py
b/airflow-core/src/airflow/api_fastapi/core_api/routes/public/plugins.py
index 57c27a99b68..f9df579f929 100644
--- a/airflow-core/src/airflow/api_fastapi/core_api/routes/public/plugins.py
+++ b/airflow-core/src/airflow/api_fastapi/core_api/routes/public/plugins.py
@@ -22,6 +22,7 @@ from fastapi import Depends
from pydantic import ValidationError
from airflow import plugins_manager
+from airflow.api_fastapi.app import get_auth_manager
from airflow.api_fastapi.auth.managers.models.resource_details import
AccessView
from airflow.api_fastapi.common.parameters import QueryLimit, QueryOffset
from airflow.api_fastapi.common.router import AirflowRouter
@@ -30,7 +31,8 @@ from airflow.api_fastapi.core_api.datamodels.plugins import (
PluginImportErrorCollectionResponse,
PluginResponse,
)
-from airflow.api_fastapi.core_api.security import requires_access_view
+from airflow.api_fastapi.core_api.security import GetUserDep,
requires_access_view
+from airflow.configuration import conf
logger = structlog.get_logger(__name__)
@@ -44,8 +46,20 @@ plugins_router = AirflowRouter(tags=["Plugin"],
prefix="/plugins")
def get_plugins(
limit: QueryLimit,
offset: QueryOffset,
+ user: GetUserDep,
) -> PluginCollectionResponse:
plugins_info = sorted(plugins_manager.get_plugin_info(), key=lambda x:
x["name"])
+
+ # In multi-team mode, a team-scoped plugin is only visible to users
authorized for
+ # its team. Global plugins (team_name is None) remain visible to everyone.
+ if conf.getboolean("core", "multi_team"):
+ authorized_teams = get_auth_manager().get_authorized_teams(user=user)
+ plugins_info = [
+ plugin_dict
+ for plugin_dict in plugins_info
+ if plugin_dict.get("team_name") is None or
plugin_dict["team_name"] in authorized_teams
+ ]
+
valid_plugins: list[PluginResponse] = []
for plugin_dict in plugins_info:
try:
diff --git a/airflow-core/src/airflow/plugins_manager.py
b/airflow-core/src/airflow/plugins_manager.py
index 2aa93951e89..f747e8d37f9 100644
--- a/airflow-core/src/airflow/plugins_manager.py
+++ b/airflow-core/src/airflow/plugins_manager.py
@@ -366,7 +366,7 @@ def get_plugin_info(attrs_to_dump: Iterable[str] | None =
None) -> list[dict[str
}
plugins_info = []
for plugin in _get_plugins()[0]:
- info: dict[str, Any] = {"name": plugin.name}
+ info: dict[str, Any] = {"name": plugin.name, "team_name":
plugin.team_name}
for attr in attrs_to_dump:
if attr in ("global_operator_extra_links", "operator_extra_links"):
info[attr] = [f"<{qualname(d.__class__)} object>" for d in
getattr(plugin, attr)]
@@ -422,3 +422,40 @@ def get_priority_weight_strategy_plugins() -> dict[str,
type[PriorityWeightStrat
def get_import_errors() -> dict[str, str]:
"""Get import errors encountered during plugin loading."""
return _get_plugins()[1]
+
+
+def validate_plugin_teams() -> None:
+ """
+ Validate that every team-scoped plugin references a team that exists in
the database.
+
+ Only enforced when multi-team mode is enabled. This must run in a context
with
+ metadata database access (the API server) — never in the Dag processor,
triggerer,
+ or workers, which reach the database only through the Execution API.
+
+ Raises ``AirflowConfigException`` if any plugin declares a ``team_name``
that is not
+ present in the database, naming the offending plugins and unknown teams so
the
+ misconfiguration can be fixed quickly.
+ """
+ if not conf.getboolean("core", "multi_team"):
+ return
+
+ from airflow.exceptions import AirflowConfigException
+ from airflow.models.team import Team
+
+ known_teams = Team.get_all_team_names()
+ unknown_by_plugin = {
+ plugin.name: plugin.team_name
+ for plugin in _get_plugins()[0]
+ if plugin.team_name is not None and plugin.team_name not in known_teams
+ }
+ if unknown_by_plugin:
+ lines = [
+ f" - Plugin '{plugin_name}' is assigned to team '{team_name}',
which does not exist."
+ for plugin_name, team_name in unknown_by_plugin.items()
+ ]
+ raise AirflowConfigException(
+ "Some plugins are assigned to teams that do not exist:\n"
+ + "\n".join(lines)
+ + "\n\nCreate a team with `airflow teams create <team_name>`, "
+ "or update the plugin to use an existing team."
+ )
diff --git a/airflow-core/src/airflow/ui/openapi-gen/requests/schemas.gen.ts
b/airflow-core/src/airflow/ui/openapi-gen/requests/schemas.gen.ts
index e1880fdd4ea..a2fc21293c8 100644
--- a/airflow-core/src/airflow/ui/openapi-gen/requests/schemas.gen.ts
+++ b/airflow-core/src/airflow/ui/openapi-gen/requests/schemas.gen.ts
@@ -5802,6 +5802,17 @@ export const $PluginResponse = {
type: 'string',
title: 'Name'
},
+ team_name: {
+ anyOf: [
+ {
+ type: 'string'
+ },
+ {
+ type: 'null'
+ }
+ ],
+ title: 'Team Name'
+ },
macros: {
items: {
type: 'string'
diff --git a/airflow-core/src/airflow/ui/openapi-gen/requests/types.gen.ts
b/airflow-core/src/airflow/ui/openapi-gen/requests/types.gen.ts
index 24b1fbe5f2e..fa719ec13df 100644
--- a/airflow-core/src/airflow/ui/openapi-gen/requests/types.gen.ts
+++ b/airflow-core/src/airflow/ui/openapi-gen/requests/types.gen.ts
@@ -1534,6 +1534,7 @@ export type PluginImportErrorResponse = {
*/
export type PluginResponse = {
name: string;
+ team_name?: string | null;
macros: Array<(string)>;
flask_blueprints: Array<(string)>;
fastapi_apps: Array<FastAPIAppResponse>;
diff --git
a/airflow-core/tests/unit/api_fastapi/core_api/routes/public/test_plugins.py
b/airflow-core/tests/unit/api_fastapi/core_api/routes/public/test_plugins.py
index e995cb4b137..1d9ba2175fa 100644
--- a/airflow-core/tests/unit/api_fastapi/core_api/routes/public/test_plugins.py
+++ b/airflow-core/tests/unit/api_fastapi/core_api/routes/public/test_plugins.py
@@ -16,12 +16,17 @@
# under the License.
from __future__ import annotations
+from unittest import mock
from unittest.mock import patch
import pytest
+from airflow.plugins_manager import AirflowPlugin
+
from tests_common.test_utils.asserts import assert_queries_count
+from tests_common.test_utils.config import conf_vars
from tests_common.test_utils.markers import
skip_if_force_lowest_dependencies_marker
+from tests_common.test_utils.mock_plugins import mock_plugin_manager
pytestmark = pytest.mark.db_test
@@ -201,3 +206,71 @@ class TestGetPluginImportErrors:
def test_should_response_403(self, unauthorized_test_client):
response = unauthorized_test_client.get("/plugins/importErrors")
assert response.status_code == 403
+
+
+class _GlobalPlugin(AirflowPlugin):
+ name = "global_plugin"
+
+
+class _TeamAPlugin(AirflowPlugin):
+ name = "team_a_plugin"
+ team_name = "team_a"
+
+
+class _TeamBPlugin(AirflowPlugin):
+ name = "team_b_plugin"
+ team_name = "team_b"
+
+
+@skip_if_force_lowest_dependencies_marker
+class TestGetPluginsTeamFiltering:
+ _plugins = [_GlobalPlugin(), _TeamAPlugin(), _TeamBPlugin()]
+
+ def test_no_filtering_when_multi_team_disabled(self, test_client):
+ """With multi_team off, all plugins are returned regardless of
team_name."""
+ with mock_plugin_manager(plugins=self._plugins):
+ response = test_client.get("/plugins")
+
+ assert response.status_code == 200
+ names = {plugin["name"] for plugin in response.json()["plugins"]}
+ assert names == {"global_plugin", "team_a_plugin", "team_b_plugin"}
+
+ @conf_vars({("core", "multi_team"): "True"})
+
@mock.patch("airflow.api_fastapi.core_api.routes.public.plugins.get_auth_manager")
+ def test_filters_to_authorized_teams_and_global(self,
mock_get_auth_manager, test_client):
+ """A user authorized for team_a sees global + team_a plugins, but not
team_b."""
+ mock_get_auth_manager.return_value.get_authorized_teams.return_value =
{"team_a"}
+
+ with mock_plugin_manager(plugins=self._plugins):
+ response = test_client.get("/plugins")
+
+ assert response.status_code == 200
+ body = response.json()
+ names = {plugin["name"] for plugin in body["plugins"]}
+ assert names == {"global_plugin", "team_a_plugin"}
+ assert body["total_entries"] == 2
+
+ @conf_vars({("core", "multi_team"): "True"})
+
@mock.patch("airflow.api_fastapi.core_api.routes.public.plugins.get_auth_manager")
+ def test_user_with_no_teams_sees_only_global(self, mock_get_auth_manager,
test_client):
+ mock_get_auth_manager.return_value.get_authorized_teams.return_value =
set()
+
+ with mock_plugin_manager(plugins=self._plugins):
+ response = test_client.get("/plugins")
+
+ assert response.status_code == 200
+ body = response.json()
+ names = {plugin["name"] for plugin in body["plugins"]}
+ assert names == {"global_plugin"}
+ assert body["total_entries"] == 1
+
+ @conf_vars({("core", "multi_team"): "True"})
+
@mock.patch("airflow.api_fastapi.core_api.routes.public.plugins.get_auth_manager")
+ def test_team_name_present_in_response(self, mock_get_auth_manager,
test_client):
+ mock_get_auth_manager.return_value.get_authorized_teams.return_value =
{"team_a"}
+
+ with mock_plugin_manager(plugins=self._plugins):
+ response = test_client.get("/plugins")
+
+ team_name_by_plugin = {p["name"]: p["team_name"] for p in
response.json()["plugins"]}
+ assert team_name_by_plugin == {"global_plugin": None, "team_a_plugin":
"team_a"}
diff --git a/airflow-core/tests/unit/cli/commands/test_plugins_command.py
b/airflow-core/tests/unit/cli/commands/test_plugins_command.py
index 8a6944483a8..57070c6cd8c 100644
--- a/airflow-core/tests/unit/cli/commands/test_plugins_command.py
+++ b/airflow-core/tests/unit/cli/commands/test_plugins_command.py
@@ -79,6 +79,7 @@ class TestPluginsCommand:
assert info == [
{
"name": "test_plugin",
+ "team_name": None,
"admin_views": [],
"macros": ["unit.plugins.test_plugin.plugin_macro"],
"menu_links": [],
diff --git a/airflow-core/tests/unit/plugins/test_plugins_manager.py
b/airflow-core/tests/unit/plugins/test_plugins_manager.py
index 0dcac96fe0a..cccb8d4228b 100644
--- a/airflow-core/tests/unit/plugins/test_plugins_manager.py
+++ b/airflow-core/tests/unit/plugins/test_plugins_manager.py
@@ -433,3 +433,71 @@ class TestWindowPluginRegistration:
assert qualname(MyCustomWindow) in registered
assert registered[qualname(MyCustomWindow)] is MyCustomWindow
+
+
+class TestPluginTeamName:
+ """``team_name`` exposure through ``get_plugin_info`` (attribute default
is covered
+ by the shared plugins_manager tests)."""
+
+ def test_get_plugin_info_includes_team_name(self):
+ from airflow import plugins_manager
+
+ class GlobalPlugin(AirflowPlugin):
+ name = "global_plugin"
+
+ class TeamPlugin(AirflowPlugin):
+ name = "team_plugin"
+ team_name = "team_a"
+
+ with mock_plugin_manager(plugins=[GlobalPlugin(), TeamPlugin()]):
+ info_by_name = {info["name"]: info for info in
plugins_manager.get_plugin_info()}
+
+ assert info_by_name["global_plugin"]["team_name"] is None
+ assert info_by_name["team_plugin"]["team_name"] == "team_a"
+
+
+class TestValidatePluginTeams:
+ """``validate_plugin_teams`` startup validation."""
+
+ def test_no_op_when_multi_team_disabled(self):
+ from airflow import plugins_manager
+
+ class TeamPlugin(AirflowPlugin):
+ name = "team_plugin"
+ team_name = "nonexistent_team"
+
+ # multi_team defaults to False; validation must return early without
hitting
+ # the database, even for a plugin pointing at a nonexistent team.
+ with mock_plugin_manager(plugins=[TeamPlugin()]):
+ with mock.patch("airflow.models.team.Team.get_all_team_names") as
mock_get_all_team_names:
+ plugins_manager.validate_plugin_teams()
+ mock_get_all_team_names.assert_not_called()
+
+ @conf_vars({("core", "multi_team"): "True"})
+ @mock.patch("airflow.models.team.Team.get_all_team_names",
return_value={"team_a"})
+ def test_passes_for_global_and_known_team_plugins(self,
mock_get_all_team_names):
+ from airflow import plugins_manager
+
+ class GlobalPlugin(AirflowPlugin):
+ name = "global_plugin"
+
+ class TeamPlugin(AirflowPlugin):
+ name = "team_plugin"
+ team_name = "team_a"
+
+ with mock_plugin_manager(plugins=[GlobalPlugin(), TeamPlugin()]):
+ plugins_manager.validate_plugin_teams()
+
+ @conf_vars({("core", "multi_team"): "True"})
+ @mock.patch("airflow.models.team.Team.get_all_team_names",
return_value={"team_a"})
+ def test_raises_for_unknown_team(self, mock_get_all_team_names):
+ from airflow import plugins_manager
+ from airflow.exceptions import AirflowConfigException
+
+ class TeamPlugin(AirflowPlugin):
+ name = "team_plugin"
+ team_name = "unknown_team"
+
+ with mock_plugin_manager(plugins=[TeamPlugin()]):
+ with pytest.raises(AirflowConfigException,
match="team_plugin.*unknown_team"):
+ plugins_manager.validate_plugin_teams()
diff --git a/airflow-ctl/src/airflowctl/api/datamodels/generated.py
b/airflow-ctl/src/airflowctl/api/datamodels/generated.py
index 9bb75d8dd87..5b89aca9712 100644
--- a/airflow-ctl/src/airflowctl/api/datamodels/generated.py
+++ b/airflow-ctl/src/airflowctl/api/datamodels/generated.py
@@ -2000,6 +2000,7 @@ class PluginResponse(BaseModel):
"""
name: Annotated[str, Field(title="Name")]
+ team_name: Annotated[str | None, Field(title="Team Name")] = None
macros: Annotated[list[str], Field(title="Macros")]
flask_blueprints: Annotated[list[str], Field(title="Flask Blueprints")]
fastapi_apps: Annotated[list[FastAPIAppResponse], Field(title="Fastapi
Apps")]
diff --git
a/shared/plugins_manager/src/airflow_shared/plugins_manager/plugins_manager.py
b/shared/plugins_manager/src/airflow_shared/plugins_manager/plugins_manager.py
index e8dea0d1294..01f9752718e 100644
---
a/shared/plugins_manager/src/airflow_shared/plugins_manager/plugins_manager.py
+++
b/shared/plugins_manager/src/airflow_shared/plugins_manager/plugins_manager.py
@@ -89,6 +89,14 @@ class AirflowPlugin:
"""Class used to define AirflowPlugin."""
name: str | None = None
+
+ # The team that owns this plugin when multi-team mode is enabled.
+ # ``None`` means the plugin is global: its listeners, macros, API
endpoints, and
+ # UI views are available to all teams (the default, backwards-compatible
behavior).
+ # When set, the plugin's behavior is scoped to the named team. Set at
code/build
+ # time in the plugin definition; there is no deployment-time override.
+ team_name: str | None = None
+
source: AirflowPluginSource | None = None
macros: list[Any] = []
admin_views: list[Any] = []
diff --git
a/shared/plugins_manager/tests/plugins_manager/test_plugins_manager.py
b/shared/plugins_manager/tests/plugins_manager/test_plugins_manager.py
index 723c0ae4b54..414259908ae 100644
--- a/shared/plugins_manager/tests/plugins_manager/test_plugins_manager.py
+++ b/shared/plugins_manager/tests/plugins_manager/test_plugins_manager.py
@@ -25,6 +25,7 @@ from unittest import mock
import pytest
from airflow_shared.plugins_manager import (
+ AirflowPlugin,
EntryPointSource,
PluginsDirectorySource,
_load_entrypoint_plugins,
@@ -109,3 +110,18 @@ class TestPluginsManager:
"test.plugins.test_plugins_manager",
"my_fake_module not found",
) in import_errors.items()
+
+
+class TestAirflowPluginTeamName:
+ def test_team_name_defaults_to_none(self):
+ class GlobalPlugin(AirflowPlugin):
+ name = "global_plugin"
+
+ assert GlobalPlugin.team_name is None
+
+ def test_team_name_can_be_scoped_to_a_team(self):
+ class TeamPlugin(AirflowPlugin):
+ name = "team_plugin"
+ team_name = "team_a"
+
+ assert TeamPlugin.team_name == "team_a"
