Airflow 3: python operators deprecations removal

airflow/operators/python.py

@@ -77,38 +77,6 @@ def is_venv_installed() -> bool:
     return False
 
 
-def task(python_callable: Callable | None = None, multiple_outputs: bool | None = None, **kwargs):
-    """
-    Use :func:`airflow.decorators.task` instead, this is deprecated.
-
-    Calls ``@task.python`` and allows users to turn a Python function into
-    an Airflow task.
-
-    :param python_callable: A reference to an object that is callable
-    :param op_kwargs: a dictionary of keyword arguments that will get unpacked
-        in your function (templated)
-    :param op_args: a list of positional arguments that will get unpacked when
-        calling your callable (templated)
-    :param multiple_outputs: if set, function return value will be
-        unrolled to multiple XCom values. Dict will unroll to xcom values with keys as keys.
-        Defaults to False.
-    """
-    # To maintain backwards compatibility, we import the task object into this file
-    # This prevents breakages in dags that use `from airflow.operators.python import task`
-    from airflow.decorators.python import python_task
-
-    warnings.warn(
-        """airflow.operators.python.task is deprecated. Please use the following instead
-
-        from airflow.decorators import task
-        @task
-        def my_task()""",
-        RemovedInAirflow3Warning,
-        stacklevel=2,
-    )
-    return python_task(python_callable=python_callable, multiple_outputs=multiple_outputs, **kwargs)
-
-
 @cache
 def _parse_version_info(text: str) -> tuple[int, int, int, str, int]:
     """Parse python version info from a text."""
@@ -213,13 +181,6 @@ def __init__(
         show_return_value_in_logs: bool = True,
         **kwargs,
     ) -> None:
-        if kwargs.get("provide_context"):
-            warnings.warn(
-                "provide_context is deprecated as of 2.0 and is no longer required",
-                RemovedInAirflow3Warning,
-                stacklevel=2,
-            )
-            kwargs.pop("provide_context", None)
         super().__init__(**kwargs)
         if not callable(python_callable):
             raise AirflowException("`python_callable` param must be callable")
@@ -732,11 +693,8 @@ def __init__(
                 f"Sys version: {sys.version_info}. Virtual environment version: {python_version}"
             )
         if python_version is not None and not isinstance(python_version, str):
-            warnings.warn(
-                "Passing non-string types (e.g. int or float) as python_version "
-                "is deprecated. Please use string value instead.",
-                RemovedInAirflow3Warning,
-                stacklevel=2,
+            raise AirflowException(
+                "Passing non-string types (e.g. int or float) as python_version not supported"
             )
         if not is_venv_installed():
             raise AirflowException("PythonVirtualenvOperator requires virtualenv, please install it.")

newsfragments/41493.significant.rst

@@ -0,0 +1 @@
+Removed deprecated method ``task`` from ``airflow.operators.python`` operator. Please use ``python_task`` from ``airflow.decorators.python`` instead.

tests/operators/test_python.py

@@ -292,7 +292,8 @@ def func(custom, dag):
             assert 1 == custom, "custom should be 1"
             assert dag is not None, "dag should be set"
 
-        with pytest.warns(RemovedInAirflow3Warning):
+        error_message = "Invalid arguments were passed to PythonOperator \\(task_id: task_test-provide-context-does-not-fail\\). Invalid arguments were:\n\\*\\*kwargs: {'provide_context': True}"
+        with pytest.raises(AirflowException, match=error_message):
             self.run_as_task(func, op_kwargs={"custom": 1}, provide_context=True)
 
     def test_context_with_conflicting_op_args(self):
@@ -1334,10 +1335,7 @@ def f():
                 return
             raise RuntimeError
 
-        with pytest.warns(
-            RemovedInAirflow3Warning, match="Passing non-string types.*python_version is deprecated"
-        ):
-            self.run_as_task(f, python_version=3, serializer=serializer, requirements=extra_requirements)
+        self.run_as_task(f, python_version="3", serializer=serializer, requirements=extra_requirements)
 
     def test_with_default(self):
         def f(a):