This is an automated email from the ASF dual-hosted git repository.
onikolas 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 4e8bfe8dd89 Update multi-team docs for per team executor config
(#62545)
4e8bfe8dd89 is described below
commit 4e8bfe8dd8928a4d221cf91b6a27e53a18d22003
Author: Niko Oliveira <[email protected]>
AuthorDate: Sat Feb 28 09:08:31 2026 -0800
Update multi-team docs for per team executor config (#62545)
* Update multi-team docs for per team executor config
Show detailed instructions for how to configure executors per team.
Also explicitly avoid cmd and secret config lookup. As it is not yet
supported for the 3.2 experimental release.
* Clarify how team executor config falls back to defaults
---
airflow-core/docs/core-concepts/multi-team.rst | 78 ++++++++++++++
.../src/airflow_shared/configuration/parser.py | 6 ++
.../tests/configuration/test_parser.py | 119 +++++++++++++++++++++
3 files changed, 203 insertions(+)
diff --git a/airflow-core/docs/core-concepts/multi-team.rst
b/airflow-core/docs/core-concepts/multi-team.rst
index f7b84dcbc41..65bc0af245f 100644
--- a/airflow-core/docs/core-concepts/multi-team.rst
+++ b/airflow-core/docs/core-concepts/multi-team.rst
@@ -283,6 +283,83 @@ Example configurations:
# Invalid: Duplicate Executor within a Team
executor =
LocalExecutor;team_a=CeleryExecutor,CeleryExecutor;team_b=LocalExecutor
+Team-specific Executor Settings
+^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
+
+When multiple teams use the same executor type (e.g., both ``team_a`` and
``team_b`` using ``CeleryExecutor``),
+each team can provide its own configuration for that executor. This allows
teams to point to different Celery
+brokers, use different Kubernetes namespaces, or customize any executor
setting independently.
+
+Configuration Resolution Order
+"""""""""""""""""""""""""""""""
+
+When a team executor reads a configuration value (e.g., ``[celery]
broker_url``), the system checks the
+following sources in order, returning the first value found:
+
+1. **Team-specific environment variable** —
``AIRFLOW__{TEAM}___{SECTION}__{KEY}``
+2. **Team-specific config file section** — ``[team_name=section]``
+3. **Default values** — built-in defaults or ``fallback`` values
+
+The following sources are **skipped** for team executors (they do not yet
support team-based configuration):
+
+- **Command execution** (``{key}_cmd``)
+- **Secrets backend** (``{key}_secret``)
+
+.. note::
+
+ Team-specific configuration does **not** fall back to the global
environment variable or global config file
+ settings. For example, if there is a global ``CeleryExecutor`` and a team
``CeleryExecutor`` in use, the global
+ ``CeleryExecutor`` may want to increase ``celery.worker_concurrency`` from
the default of ``16`` to ``32`` by
+ overriding this configuration. However, the team ``CeleryExecutor``
should not be forced to ``32``, it will
+ continue to use the default of ``16`` unless it is explicitly overridden
with team-specific configuration.
+
+Via Environment Variables
+"""""""""""""""""""""""""
+
+Team-specific configuration can be provided using environment variables with
the following format:
+
+.. code-block:: text
+
+ AIRFLOW__{TEAM}___{SECTION}__{KEY}
+
+Note the delimiters: double underscore before the team name (part of the
``AIRFLOW__`` prefix), **triple
+underscore** between the team name and section, and double underscore between
section and key. The team name
+is uppercase.
+
+.. code-block:: bash
+
+ # team_a's Celery broker URL
+ export AIRFLOW__TEAM_A___CELERY__BROKER_URL="redis://team-a-redis:6379/0"
+
+ # team_b's Celery broker URL
+ export AIRFLOW__TEAM_B___CELERY__BROKER_URL="redis://team-b-redis:6379/0"
+
+ # team_b's Celery result backend
+ export
AIRFLOW__TEAM_B___CELERY__RESULT_BACKEND="db+postgresql://team-b-db/celery_results"
+
+Via Config File
+"""""""""""""""
+
+Team-specific settings can also be placed in the ``airflow.cfg`` file using
sections prefixed with the team
+name followed by an equals sign:
+
+.. code-block:: ini
+
+ # Global celery settings (used by the global executor, NOT as a fallback
for teams)
+ [celery]
+ broker_url = redis://default-redis:6379/0
+ result_backend = db+postgresql://default-db/celery_results
+
+ # team_a overrides
+ [team_a=celery]
+ broker_url = redis://team-a-redis:6379/0
+ result_backend = db+postgresql://team-a-db/celery_results
+
+ # team_b overrides
+ [team_b=celery]
+ broker_url = redis://team-b-redis:6379/0
+ result_backend = db+postgresql://team-b-db/celery_results
+
Dag Bundle to Team Association
------------------------------
@@ -370,6 +447,7 @@ Multi-Team mode is currently an experimental feature in
preview. It is not yet f
- Async support (Triggers, Event Driven Scheduling, async Callbacks, etc)
- Some UI elements may not be fully team-aware
- Full provider support for executors and secrets backends
+- Command and Secrets based lookup for team based configuration
- Plugins
Global Uniqueness of Identifiers
diff --git a/shared/configuration/src/airflow_shared/configuration/parser.py
b/shared/configuration/src/airflow_shared/configuration/parser.py
index cd2e6c6511e..455a6282924 100644
--- a/shared/configuration/src/airflow_shared/configuration/parser.py
+++ b/shared/configuration/src/airflow_shared/configuration/parser.py
@@ -886,6 +886,9 @@ class AirflowConfigParser(ConfigParser):
**kwargs,
) -> str | ValueNotFound:
"""Get config option from command execution."""
+ if kwargs.get("team_name", None):
+ # Commands based team config fetching is not currently supported
+ return VALUE_NOT_FOUND_SENTINEL
option = self._get_cmd_option(section, key)
if option:
return option
@@ -909,6 +912,9 @@ class AirflowConfigParser(ConfigParser):
**kwargs,
) -> str | ValueNotFound:
"""Get config option from secrets backend."""
+ if kwargs.get("team_name", None):
+ # Secrets based team config fetching is not currently supported
+ return VALUE_NOT_FOUND_SENTINEL
option = self._get_secret_option(section, key)
if option:
return option
diff --git a/shared/configuration/tests/configuration/test_parser.py
b/shared/configuration/tests/configuration/test_parser.py
index 16c8667dfad..a018a931ed5 100644
--- a/shared/configuration/tests/configuration/test_parser.py
+++ b/shared/configuration/tests/configuration/test_parser.py
@@ -888,3 +888,122 @@ existing_list = one,two,three
# case 3: active (non-deprecated) key
# Active key should be present
assert test_conf.get("test_section", "active_key") == "active_value"
+
+ def test_team_env_var_takes_priority(self):
+ """Test that team-specific env var is returned when team_name is
provided."""
+ test_config = textwrap.dedent(
+ """\
+ [celery]
+ broker_url = redis://global:6379/0
+ """
+ )
+ test_conf = AirflowConfigParser(default_config=test_config)
+ with patch.dict(
+ os.environ,
+ {"AIRFLOW__TEAM_A___CELERY__BROKER_URL": "redis://team-a:6379/0"},
+ ):
+ assert test_conf.get("celery", "broker_url", team_name="team_a")
== "redis://team-a:6379/0"
+
+ def test_team_config_file_section(self):
+ """Test that [team_name=section] in config file is used when team_name
is provided."""
+ test_conf = AirflowConfigParser()
+ test_conf.read_string(
+ textwrap.dedent(
+ """\
+ [celery]
+ broker_url = redis://global:6379/0
+
+ [team_a=celery]
+ broker_url = redis://team-a:6379/0
+ """
+ )
+ )
+ assert test_conf.get("celery", "broker_url", team_name="team_a") ==
"redis://team-a:6379/0"
+
+ def test_team_does_not_fallback_to_global_config(self):
+ """Test that team lookup does NOT fall back to global config section
or env var."""
+ test_conf = AirflowConfigParser()
+ test_conf.read_string(
+ textwrap.dedent(
+ """\
+ [celery]
+ broker_url = redis://global:6379/0
+ """
+ )
+ )
+ # team_a has no config set, should NOT get global value; should fall
through to defaults
+ with pytest.raises(AirflowConfigException):
+ test_conf.get("celery", "broker_url", team_name="team_a")
+
+ def test_team_does_not_fallback_to_global_env_var(self):
+ """Test that team lookup does NOT fall back to global env var."""
+ test_conf = AirflowConfigParser()
+ with patch.dict(os.environ, {"AIRFLOW__CELERY__BROKER_URL":
"redis://global-env:6379/0"}):
+ with pytest.raises(AirflowConfigException):
+ test_conf.get("celery", "broker_url", team_name="team_a")
+
+ def test_team_skips_cmd_lookup(self):
+ """Test that _cmd config values are skipped when team_name is
provided."""
+ test_conf = AirflowConfigParser()
+ test_conf.read_string(
+ textwrap.dedent(
+ """\
+ [test]
+ sensitive_key_cmd = echo -n cmd_value
+ """
+ )
+ )
+ test_conf.sensitive_config_values.add(("test", "sensitive_key"))
+
+ # Without team_name, cmd works
+ assert test_conf.get("test", "sensitive_key") == "cmd_value"
+
+ # With team_name, cmd is skipped
+ with pytest.raises(AirflowConfigException):
+ test_conf.get("test", "sensitive_key", team_name="team_a")
+
+ def test_team_skips_secret_lookup(self):
+ """Test that _secret config values are skipped when team_name is
provided."""
+
+ class TestParserWithSecretBackend(AirflowConfigParser):
+ def __init__(self, *args, **kwargs):
+ super().__init__(*args, **kwargs)
+ self.configuration_description = {}
+ self._default_values = ConfigParser()
+ self._suppress_future_warnings = False
+
+ def _get_config_value_from_secret_backend(self, config_key: str)
-> str | None:
+ return "secret_value_from_backend"
+
+ test_conf = TestParserWithSecretBackend()
+ test_conf.read_string(
+ textwrap.dedent(
+ """\
+ [test]
+ sensitive_key_secret = test/secret/path
+ """
+ )
+ )
+ test_conf.sensitive_config_values.add(("test", "sensitive_key"))
+
+ # Without team_name, secret backend works
+ assert test_conf.get("test", "sensitive_key") ==
"secret_value_from_backend"
+
+ # With team_name, secret backend is skipped
+ with pytest.raises(AirflowConfigException):
+ test_conf.get("test", "sensitive_key", team_name="team_a")
+
+ def test_team_falls_through_to_defaults(self):
+ """Test that team lookup falls through to defaults when no
team-specific value is set."""
+ test_conf = AirflowConfigParser()
+ # "test" section with "key1" having default "default_value" is set in
the AirflowConfigParser fixture
+ assert test_conf.get("test", "key1", team_name="team_a") ==
"default_value"
+
+ def test_team_env_var_format(self):
+ """Test the triple-underscore env var format:
AIRFLOW__{TEAM}___{SECTION}__{KEY}."""
+ test_conf = AirflowConfigParser()
+ with patch.dict(
+ os.environ,
+ {"AIRFLOW__MY_TEAM___MY_SECTION__MY_KEY": "team_value"},
+ ):
+ assert test_conf.get("my_section", "my_key", team_name="my_team")
== "team_value"
