Re-introduce TypeGuard to SecretsMasker

task-sdk/src/airflow/sdk/execution_time/secrets_masker.py

@@ -27,10 +27,17 @@
 from enum import Enum
 from functools import cache, cached_property
 from re import Pattern
-from typing import Any, TextIO, TypeAlias, TypeVar, overload
+from typing import TYPE_CHECKING, Any, Protocol, TextIO, TypeAlias, TypeVar, overload
 
 from airflow import settings
 
+if TYPE_CHECKING:
+    from typing import TypeGuard
+
+    class _V1EnvVarLike(Protocol):
+        def to_dict(self) -> dict[str, Any]: ...
+
+
 V1EnvVar = TypeVar("V1EnvVar")
 Redactable: TypeAlias = str | V1EnvVar | dict[Any, Any] | tuple[Any, ...] | list[Any]
 Redacted: TypeAlias = Redactable | str
@@ -167,13 +174,13 @@ def reset_secrets_masker() -> None:
 def _get_v1_env_var_type() -> type:
     try:
         from kubernetes.client import V1EnvVar
+
+        return V1EnvVar
     except ImportError:
         return type("V1EnvVar", (), {})
-    return V1EnvVar
 
 
-# TODO update return type to TypeGuard[V1EnvVar] once mypy 1.17.0 is available
-def _is_v1_env_var(v: Any) -> bool:
+def _is_v1_env_var(v: Any) -> TypeGuard[_V1EnvVarLike]:
     return isinstance(v, _get_v1_env_var_type())
 
 
@@ -275,8 +282,8 @@ def _redact(self, item: Redactable, name: str | None, depth: int, max_depth: int
                 return to_return
             if isinstance(item, Enum):
                 return self._redact(item=item.value, name=name, depth=depth, max_depth=max_depth)
-            if _is_v1_env_var(item) and hasattr(item, "to_dict"):
-                tmp: dict = item.to_dict()
+            if _is_v1_env_var(item):
+                tmp = item.to_dict()
                 if should_hide_value_for_key(tmp.get("name", "")) and "value" in tmp:
                     tmp["value"] = "***"
                 else:

task-sdk/tests/task_sdk/definitions/test_secrets_masker.py

@@ -609,7 +609,10 @@ def to_dict(self):
         secrets_masker = SecretsMasker()
 
         with patch("airflow.sdk.execution_time.secrets_masker._secrets_masker", return_value=secrets_masker):
-            with patch("airflow.sdk.execution_time.secrets_masker._is_v1_env_var", return_value=True):
+            with patch(
+                "airflow.sdk.execution_time.secrets_masker._is_v1_env_var",
+                side_effect=lambda a: isinstance(a, MockV1EnvVar),
+            ):
                 redacted_secret = redact(secret_env_var)
                 redacted_normal = redact(normal_env_var)