This is an automated email from the ASF dual-hosted git repository.
kaxil 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 ea955744e4d Make airflow standalone discover new Dag files within
seconds (#74479)
ea955744e4d is described below
commit ea955744e4dc735cdfa874f64e8bf4c9bc04d174
Author: Kaxil Naik <[email protected]>
AuthorDate: Fri Oct 9 16:22:32 2026 +0100
Make airflow standalone discover new Dag files within seconds (#74479)
A Dag file added after startup is only found when its bundle refreshes, and
[dag_processor] refresh_interval defaults to 300 seconds, so under
airflow standalone a new file took up to five minutes to show up.
Standalone now passes its components a refresh_interval equal to
[dag_processor] bundle_refresh_check_interval (5 seconds by default), so the
Dags folder is rescanned each time the Dag processor checks whether a bundle
is due. It does this only while refresh_interval and dag_bundle_config_list
are at their defaults and the interval is not set in the environment, so an
explicit choice wins and remote bundles keep their fetch cadence.
---
.../src/airflow/cli/commands/standalone_command.py | 31 +++++++++++++
.../src/airflow/config_templates/config.yml | 4 ++
.../unit/cli/commands/test_standalone_command.py | 51 ++++++++++++++++++++++
3 files changed, 86 insertions(+)
diff --git a/airflow-core/src/airflow/cli/commands/standalone_command.py
b/airflow-core/src/airflow/cli/commands/standalone_command.py
index 943b85d53f5..15a9f5edd1d 100644
--- a/airflow-core/src/airflow/cli/commands/standalone_command.py
+++ b/airflow-core/src/airflow/cli/commands/standalone_command.py
@@ -16,6 +16,7 @@
# under the License.
from __future__ import annotations
+import json
import logging
import os
import socket
@@ -44,6 +45,8 @@ if TYPE_CHECKING:
FORCE_COLOR = bool(os.environ.get("FORCE_COLOR", ""))
NO_COLOR = bool(os.environ.get("NO_COLOR", ""))
+REFRESH_INTERVAL_ENV_VAR = "AIRFLOW__DAG_PROCESSOR__REFRESH_INTERVAL"
+
class StandaloneCommand:
"""
@@ -191,8 +194,36 @@ class StandaloneCommand:
env["AIRFLOW__CORE__AUTH_MANAGER"] = simple_auth_manager_classpath
os.environ["AIRFLOW__CORE__AUTH_MANAGER"] =
simple_auth_manager_classpath # also in this process!
+ # A new Dag file is found only when its bundle refreshes, every few
minutes by default. Have the
+ # Dag processor refresh the Dags folder each time it checks whether
any bundle is due.
+ if self.uses_default_dag_discovery():
+ refresh_interval = conf.getint("dag_processor",
"bundle_refresh_check_interval")
+ self.print_output(
+ "standalone",
+ f"Checking for new Dag files every {refresh_interval} seconds "
+ "(set [dag_processor] refresh_interval to change this)",
+ )
+ env[REFRESH_INTERVAL_ENV_VAR] = str(refresh_interval)
+
return env
+ def uses_default_dag_discovery(self) -> bool:
+ """
+ Check that the user left the Dags folder bundle and its refresh
interval at their defaults.
+
+ The generated airflow.cfg writes out every default, so a value equal
to the default counts as
+ unset there; an environment variable counts as a choice whatever its
value. Other bundles are
+ left alone, since refreshing a remote bundle fetches from it.
+ """
+ if REFRESH_INTERVAL_ENV_VAR in os.environ:
+ return False
+ if conf.get("dag_processor", "refresh_interval") !=
conf.get_default_value(
+ "dag_processor", "refresh_interval"
+ ):
+ return False
+ default_bundles = json.loads(conf.get_default_value("dag_processor",
"dag_bundle_config_list"))
+ return conf.getjson("dag_processor", "dag_bundle_config_list") ==
default_bundles
+
def find_user_info(self):
if conf.get("core", "simple_auth_manager_all_admins").lower() ==
"true":
# If we have no auth anyways, no need to print or do anything
diff --git a/airflow-core/src/airflow/config_templates/config.yml
b/airflow-core/src/airflow/config_templates/config.yml
index 25cffae10c3..7581497d0c5 100644
--- a/airflow-core/src/airflow/config_templates/config.yml
+++ b/airflow-core/src/airflow/config_templates/config.yml
@@ -3317,6 +3317,10 @@ dag_processor:
refresh_interval:
description: |
How often (in seconds) to refresh, or look for new files, in a DAG
bundle.
+
+ When this option and ``dag_bundle_config_list`` are left at their
defaults, ``airflow standalone``
+ refreshes the Dags folder every ``bundle_refresh_check_interval``
seconds instead, so new Dag
+ files show up within seconds.
version_added: ~
type: integer
example: ~
diff --git a/airflow-core/tests/unit/cli/commands/test_standalone_command.py
b/airflow-core/tests/unit/cli/commands/test_standalone_command.py
index cd87c16ee49..5cfed400c43 100644
--- a/airflow-core/tests/unit/cli/commands/test_standalone_command.py
+++ b/airflow-core/tests/unit/cli/commands/test_standalone_command.py
@@ -17,6 +17,7 @@
# under the License.
from __future__ import annotations
+import json
import os
import subprocess
from collections import deque
@@ -26,6 +27,7 @@ from unittest import mock
import pytest
from airflow.cli.commands.standalone_command import StandaloneCommand,
SubCommand
+from airflow.configuration import conf
from airflow.executors import executor_loader
from airflow.executors.executor_constants import (
CELERY_EXECUTOR,
@@ -36,6 +38,14 @@ from airflow.executors.executor_constants import (
from tests_common.test_utils.config import conf_vars
+def default_dag_discovery_config():
+ """Dag discovery options at their defaults, as the generated airflow.cfg
writes them."""
+ return {
+ ("dag_processor", key): conf.get_default_value("dag_processor", key)
+ for key in ("refresh_interval", "dag_bundle_config_list")
+ }
+
+
class TestStandaloneCommand:
@pytest.mark.parametrize(
"conf_executor_name",
@@ -85,6 +95,47 @@ class TestStandaloneCommand:
assert "AIRFLOW__CORE__AUTH_MANAGER" not in env
+ def
test_calculate_env_refreshes_dags_folder_on_every_bundle_check_by_default(self):
+ """The generated airflow.cfg spells out every default, which must not
count as a user choice."""
+ config = {**default_dag_discovery_config(), ("dag_processor",
"bundle_refresh_check_interval"): "5"}
+ with conf_vars(config), mock.patch.dict(os.environ):
+ env = StandaloneCommand().calculate_env()
+
+ assert env["AIRFLOW__DAG_PROCESSOR__REFRESH_INTERVAL"] == "5"
+
+ @pytest.mark.parametrize(
+ ("config", "environ"),
+ [
+ pytest.param({("dag_processor", "refresh_interval"): "60"}, {},
id="interval-in-airflow-cfg"),
+ pytest.param({}, {"AIRFLOW__DAG_PROCESSOR__REFRESH_INTERVAL":
"60"}, id="interval-in-env"),
+ pytest.param(
+ {}, {"AIRFLOW__DAG_PROCESSOR__REFRESH_INTERVAL": "300"},
id="default-interval-in-env"
+ ),
+ pytest.param(
+ {
+ ("dag_processor", "dag_bundle_config_list"): json.dumps(
+ [
+ {
+ "name": "my-git-repo",
+ "classpath":
"airflow.providers.git.bundles.git.GitDagBundle",
+ "kwargs": {},
+ }
+ ]
+ )
+ },
+ {},
+ id="custom-bundles",
+ ),
+ ],
+ )
+ def test_calculate_env_keeps_user_chosen_dag_discovery(self, config,
environ):
+ with conf_vars({**default_dag_discovery_config(), **config}),
mock.patch.dict(os.environ, environ):
+ env = StandaloneCommand().calculate_env()
+
+ assert env.get("AIRFLOW__DAG_PROCESSOR__REFRESH_INTERVAL") ==
environ.get(
+ "AIRFLOW__DAG_PROCESSOR__REFRESH_INTERVAL"
+ )
+
@mock.patch("airflow.cli.commands.standalone_command.os.path.exists",
return_value=False)
@mock.patch("airflow.cli.commands.standalone_command.create_auth_manager")
@mock.patch("airflow.cli.commands.standalone_command.conf.get")
