apache · stephen-bracken · Jul 31, 2026
@@ -61,6 +61,20 @@ config:
   keycloak_auth_manager:
     description: This section contains settings for Keycloak auth manager integration.
     options:
+      cache_ttl_seconds:
+        description: |
+          Time (in seconds) to cache auth decisions in bulk ``is_authorized_{resource}`` methods.
+        type: integer
+        version_added: 0.9.0
+        example: ~
+        default: "30"
+      cache_timeout_seconds:
+        description: |
+          Timeout (in seconds) to cache auth decisions in bulk ``is_authorized_{resource}`` methods.
+        type: integer
+        version_added: 0.9.0
+        example: ~
+        default: "60"
       client_id:
         description: |
           Client ID configured in Keycloak to integrate with Airflow.

@@ -20,8 +20,15 @@
 import time
 from collections.abc import Callable
 
-_CACHE_TTL_SECONDS = 30
-_SINGLE_FLIGHT_TIMEOUT_SECONDS = 60
+from airflow.providers.common.compat.sdk import conf
+from airflow.providers.keycloak.auth_manager.constants import (
+    CONF_CACHE_TIMEOUT_SECONDS_KEY,
+    CONF_CACHE_TTL_SECONDS_KEY,
+    CONF_SECTION_NAME,
+)
+
+_CACHE_TTL_SECONDS: int | None = None
+_SINGLE_FLIGHT_TIMEOUT_SECONDS: int | None = None
 
 # Maps cache keys to (timestamp, result) pairs for TTL-based expiration.
 _cache: dict[tuple, tuple[float, frozenset[str]]] = {}
@@ -30,9 +37,25 @@
 _cache_lock = threading.Lock()
 
 
+def _cache_ttl_seconds() -> int:
+    global _CACHE_TTL_SECONDS
+    if not _CACHE_TTL_SECONDS:
+        _CACHE_TTL_SECONDS = conf.getint(CONF_SECTION_NAME, CONF_CACHE_TTL_SECONDS_KEY, fallback=30)
+    return _CACHE_TTL_SECONDS
+
+
+def _cache_timeout_seconds() -> int:
+    global _SINGLE_FLIGHT_TIMEOUT_SECONDS
+    if not _SINGLE_FLIGHT_TIMEOUT_SECONDS:
+        _SINGLE_FLIGHT_TIMEOUT_SECONDS = conf.getint(
+            CONF_SECTION_NAME, CONF_CACHE_TIMEOUT_SECONDS_KEY, fallback=60
+        )
+    return _SINGLE_FLIGHT_TIMEOUT_SECONDS
+
+
 def _cache_get(key: tuple) -> frozenset[str] | None:
     entry = _cache.get(key)
-    if entry and (time.monotonic() - entry[0]) < _CACHE_TTL_SECONDS:
+    if entry and (time.monotonic() - entry[0]) < _cache_ttl_seconds():
         return entry[1]
     return None
 
@@ -41,7 +64,7 @@ def _cache_set(key: tuple, value: frozenset[str]) -> None:
     with _cache_lock:
         _cache[key] = (time.monotonic(), value)
         now = time.monotonic()
-        for k in [k for k, (ts, _) in _cache.items() if now - ts > _CACHE_TTL_SECONDS * 2]:
+        for k in [k for k, (ts, _) in _cache.items() if now - ts > _cache_ttl_seconds() * 2]:
             _cache.pop(k, None)
 
 
@@ -67,7 +90,7 @@ def single_flight(cache_key: tuple, query_keycloak: Callable[[], set[str]]) -> s
 
     if not is_worker:
         # Wait for the other thread to finish
-        event.wait(timeout=_SINGLE_FLIGHT_TIMEOUT_SECONDS)
+        event.wait(timeout=_cache_timeout_seconds())
         cached = _cache_get(cache_key)
         if cached is not None:
             return set(cached)

@@ -19,6 +19,8 @@
 from __future__ import annotations
 
 CONF_SECTION_NAME = "keycloak_auth_manager"
+CONF_CACHE_TTL_SECONDS_KEY = "cache_ttl_seconds"
+CONF_CACHE_TIMEOUT_SECONDS_KEY = "cache_timeout_seconds"
 CONF_CLIENT_ID_KEY = "client_id"
 CONF_CLIENT_SECRET_KEY = "client_secret"
 CONF_REALM_KEY = "realm"

@@ -42,6 +42,20 @@ def get_provider_info():
             "keycloak_auth_manager": {
                 "description": "This section contains settings for Keycloak auth manager integration.",
                 "options": {
+                    "cache_ttl_seconds": {
+                        "description": "Time (in seconds) to cache auth decisions in bulk ``is_authorized_{resource}`` methods.\n",
+                        "type": "integer",
+                        "version_added": "0.9.0",
+                        "example": None,
+                        "default": "30",
+                    },
+                    "cache_timeout_seconds": {
+                        "description": "Timeout (in seconds) to cache auth decisions in bulk ``is_authorized_{resource}`` methods.\n",
+                        "type": "integer",
+                        "version_added": "0.9.0",
+                        "example": None,
+                        "default": "60",
+                    },
                     "client_id": {
                         "description": "Client ID configured in Keycloak to integrate with Airflow.\nThis client must follow the standard OpenID Connect authentication flow.\n",
                         "type": "string",

@@ -23,6 +23,13 @@
 
 from airflow.providers.keycloak.auth_manager import cache as cache_module
 from airflow.providers.keycloak.auth_manager.cache import single_flight
+from airflow.providers.keycloak.auth_manager.constants import (
+    CONF_CACHE_TIMEOUT_SECONDS_KEY,
+    CONF_CACHE_TTL_SECONDS_KEY,
+    CONF_SECTION_NAME,
+)
+
+from tests_common.test_utils.config import conf_vars
 
 
 @pytest.fixture(autouse=True)
@@ -34,7 +41,30 @@ def _clear_cache():
     cache_module._pending_requests.clear()
 
 
+@pytest.fixture(autouse=True)
+def _set_cache_ttl():
+    with conf_vars(
+        {
+            (CONF_SECTION_NAME, CONF_CACHE_TTL_SECONDS_KEY): "60",
+            (CONF_SECTION_NAME, CONF_CACHE_TIMEOUT_SECONDS_KEY): "120",
+        }
+    ):
+        yield
+
+
 class TestSingleFlight:
+    def test_configure_ttl(self):
+        assert cache_module._CACHE_TTL_SECONDS is None
+        with conf_vars({(CONF_SECTION_NAME, CONF_CACHE_TTL_SECONDS_KEY): "120"}):
+            cache_module._cache_ttl_seconds()
+            assert cache_module._CACHE_TTL_SECONDS == 120
+
+    def test_configure_timeout(self):
+        assert cache_module._SINGLE_FLIGHT_TIMEOUT_SECONDS is None
+        with conf_vars({(CONF_SECTION_NAME, CONF_CACHE_TIMEOUT_SECONDS_KEY): "120"}):
+            cache_module._cache_timeout_seconds()
+            assert cache_module._SINGLE_FLIGHT_TIMEOUT_SECONDS == 120
+
     def test_returns_query_result(self):
         result = single_flight(("key",), lambda: {"a", "b"})
         assert result == {"a", "b"}

@@ -17,6 +17,8 @@
 from __future__ import annotations
 
 from airflow.providers.keycloak.auth_manager.constants import (
+    CONF_CACHE_TIMEOUT_SECONDS_KEY,
+    CONF_CACHE_TTL_SECONDS_KEY,
     CONF_CLIENT_ID_KEY,
     CONF_CLIENT_SECRET_KEY,
     CONF_REALM_KEY,
@@ -29,6 +31,12 @@ class TestKeycloakAuthManagerConstants:
     def test_conf_section_name(self):
         assert CONF_SECTION_NAME == "keycloak_auth_manager"
 
+    def test_conf_cache_ttl_seconds_key(self):
+        assert CONF_CACHE_TTL_SECONDS_KEY == "cache_ttl_seconds"
+
+    def test_conf_cache_timeout_seconds_key(self):
+        assert CONF_CACHE_TIMEOUT_SECONDS_KEY == "cache_timeout_seconds"
+
     def test_conf_client_id_key(self):
         assert CONF_CLIENT_ID_KEY == "client_id"