This is an automated email from the ASF dual-hosted git repository.
potiuk 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 51e2c888f6f Improve is_container annotation (#58224)
51e2c888f6f is described below
commit 51e2c888f6f22208f486a6678a46cd192c76b10a
Author: Tzu-ping Chung <[email protected]>
AuthorDate: Wed Nov 12 17:25:38 2025 +0800
Improve is_container annotation (#58224)
Using TypeGuard lets the function perform type-narrowing for us, so
usages after it would not need a custom cast. The downside is we now
need to specify a bound, but we don't actually use this function that
often anyway and can just list all the possibilities.
---
airflow-core/src/airflow/utils/helpers.py | 17 +++++++++++++++--
1 file changed, 15 insertions(+), 2 deletions(-)
diff --git a/airflow-core/src/airflow/utils/helpers.py
b/airflow-core/src/airflow/utils/helpers.py
index 112b6e70c98..4d79a28d41e 100644
--- a/airflow-core/src/airflow/utils/helpers.py
+++ b/airflow-core/src/airflow/utils/helpers.py
@@ -23,7 +23,7 @@ import re
import signal
from collections.abc import Callable, Generator, Iterable, MutableMapping
from functools import cache
-from typing import TYPE_CHECKING, Any, TypeVar, cast
+from typing import TYPE_CHECKING, Any, TypeVar, cast, overload
from urllib.parse import urljoin
from lazy_object_proxy import Proxy
@@ -33,11 +33,16 @@ from airflow.exceptions import AirflowException
from airflow.utils.types import NOTSET
if TYPE_CHECKING:
+ from datetime import datetime
+ from typing import TypeGuard
+
import jinja2
from airflow.models.taskinstance import TaskInstance
from airflow.sdk.definitions.context import Context
+ CT = TypeVar("CT", str, datetime)
+
KEY_REGEX = re.compile(r"^[\w.-]+$")
GROUP_KEY_REGEX = re.compile(r"^[\w-]+$")
CAMELCASE_TO_SNAKE_CASE_REGEX = re.compile(r"(?!^)([A-Z]+)")
@@ -90,7 +95,15 @@ def prompt_with_timeout(question: str, timeout: int,
default: bool | None = None
signal.alarm(0)
-def is_container(obj: Any) -> bool:
+@overload
+def is_container(obj: None | int | Iterable[int] | range) ->
TypeGuard[Iterable[int]]: ...
+
+
+@overload
+def is_container(obj: None | CT | Iterable[CT]) -> TypeGuard[Iterable[CT]]: ...
+
+
+def is_container(obj) -> bool:
"""Test if an object is a container (iterable) but not a string."""
if isinstance(obj, Proxy):
# Proxy of any object is considered a container because it implements
__iter__
