Add __wrapped__ property to _TaskDecorator

airflow/decorators/base.py

@@ -267,6 +267,10 @@ def __call__(self, *args, **kwargs) -> XComArg:
             op.doc_md = self.function.__doc__
         return XComArg(op)
 
+    @property
+    def __wrapped__(self) -> Function:
+        return self.function
+
     @cached_property
     def function_signature(self):
         return inspect.signature(self.function)
@@ -495,6 +499,10 @@ class Task(Generic[Function]):
 
     function: Function
 
+    @property
+    def __wrapped__(self) -> Function:
+        ...
+
     def expand(self, **kwargs: "Mappable") -> XComArg:
         ...
 
@@ -527,7 +535,7 @@ def task_decorator_factory(
     **kwargs,
 ) -> TaskDecorator:
     """
-    A factory that generates a wrapper that raps a function into an Airflow operator.
+    A factory that generates a wrapper that wraps a function into an Airflow operator.
     Accepts kwargs for operator kwarg. Can be reused in a single DAG.
 
     :param python_callable: Function to decorate

tests/decorators/test_python.py

@@ -738,3 +738,20 @@ def fn(arg1, arg2):
 
     assert op.op_kwargs['arg1'] == "{{ ds }}"
     assert op.op_kwargs['arg2'] == "fn"
+
+
+def test_task_decorator_has_wrapped_attr():
+    """
+    Test  @task original underlying function is accessible
+    through the __wrapped__ attribute.
+    """
+
+    def org_test_func():
+        pass
+
+    decorated_test_func = task_decorator(org_test_func)
+
+    assert hasattr(
+        decorated_test_func, '__wrapped__'
+    ), "decorated function does not have __wrapped__ attribute"
+    assert decorated_test_func.__wrapped__ is org_test_func, "__wrapped__ attr is not the original function"