Update providers/common/compat/src/airflow/providers/common/compat/standard/operators.py

Co-authored-by: Kaxil Naik <[email protected]>

Author: 2 people

providers/common/compat/src/airflow/providers/common/compat/standard/operators.py

@@ -46,6 +46,7 @@
     from airflow.sdk.bases.operator import BaseAsyncOperator
 else:
     import inspect
+    from collections.abc import Callable
     from contextlib import suppress
     from functools import partial
 
@@ -58,7 +59,7 @@
         from airflow.models import BaseOperator
         from airflow.models.mappedoperator import OperatorPartial
 
-    def unwrap_partial(fn):
+    def unwrap_partial(fn: Callable) -> Callable:
         while isinstance(fn, partial):
             fn = fn.func
         return fn

task-sdk/src/airflow/sdk/bases/decorator.py

@@ -150,24 +150,20 @@ def _find_id_suffixes(dag: DAG) -> Iterator[int]:
     return f"{core}__{max(_find_id_suffixes(dag)) + 1}"
 
 
-def unwrap_partial(fn):
+def unwrap_partial(fn: Callable) -> Callable:
     while isinstance(fn, partial):
         fn = fn.func
     return fn
 
 
 def unwrap_callable(func):
-    from airflow.sdk.bases.decorator import _TaskDecorator
     from airflow.sdk.definitions.mappedoperator import OperatorPartial
 
-    # Airflow-specific unwrap
     if isinstance(func, (_TaskDecorator, OperatorPartial)):
         func = getattr(func, "function", getattr(func, "_func", func))
 
-    # Unwrap functools.partial
     func = unwrap_partial(func)
 
-    # Unwrap @functools.wraps chains
     with suppress(Exception):
         func = inspect.unwrap(func)