Util helper deprecated functions removal (#41520)

* util helper depreaction  removal

* Tame Mypy

* news fragment added

---------

Co-authored-by: Tzu-ping Chung <uranusjr@gmail.com>

airflow/utils/helpers.py

@@ -21,16 +21,14 @@
 import itertools
 import re
 import signal
-import warnings
 from datetime import datetime
 from functools import reduce
 from typing import TYPE_CHECKING, Any, Callable, Generator, Iterable, Mapping, MutableMapping, TypeVar, cast
 
 from lazy_object_proxy import Proxy
 
 from airflow.configuration import conf
-from airflow.exceptions import AirflowException, RemovedInAirflow3Warning
-from airflow.utils.module_loading import import_string
+from airflow.exceptions import AirflowException
 from airflow.utils.types import NOTSET
 
 if TYPE_CHECKING:
@@ -231,26 +229,6 @@ def partition(pred: Callable[[T], bool], iterable: Iterable[T]) -> tuple[Iterabl
     return itertools.filterfalse(pred, iter_1), filter(pred, iter_2)
 
 
-def chain(*args, **kwargs):
-    """Use `airflow.models.baseoperator.chain`, this function is deprecated."""
-    warnings.warn(
-        "This function is deprecated. Please use `airflow.models.baseoperator.chain`.",
-        RemovedInAirflow3Warning,
-        stacklevel=2,
-    )
-    return import_string("airflow.models.baseoperator.chain")(*args, **kwargs)
-
-
-def cross_downstream(*args, **kwargs):
-    """Use `airflow.models.baseoperator.cross_downstream`, this function is deprecated."""
-    warnings.warn(
-        "This function is deprecated. Please use `airflow.models.baseoperator.cross_downstream`.",
-        RemovedInAirflow3Warning,
-        stacklevel=2,
-    )
-    return import_string("airflow.models.baseoperator.cross_downstream")(*args, **kwargs)
-
-
 def build_airflow_url_with_query(query: dict[str, Any]) -> str:
     """
     Build airflow url using base_url and default_view and provided query.

newsfragments/41520.significant.rst

@@ -0,0 +1,5 @@
+Removed deprecated methods in airflow/utils/helpers.py
+
+ Methods removed:
+   * chain (Use ``airflow.models.baseoperator.chain``)
+   * cross_downstream (Use ``airflow.models.baseoperator.cross_downstream``)

tests/system/providers/amazon/aws/example_bedrock_retrieve_and_generate.py

@@ -35,6 +35,7 @@
 
 from airflow import DAG
 from airflow.decorators import task, task_group
+from airflow.models.baseoperator import chain
 from airflow.operators.empty import EmptyOperator
 from airflow.providers.amazon.aws.hooks.bedrock import BedrockAgentHook
 from airflow.providers.amazon.aws.hooks.opensearch_serverless import OpenSearchServerlessHook
@@ -58,7 +59,6 @@
 )
 from airflow.providers.amazon.aws.utils import get_botocore_version
 from airflow.utils.edgemodifier import Label
-from airflow.utils.helpers import chain
 from airflow.utils.trigger_rule import TriggerRule
 from tests.system.providers.amazon.aws.utils import SystemTestContextBuilder
 
@@ -450,7 +450,7 @@ def delete_opensearch_policies(collection_name: str):
 
     create_bucket = S3CreateBucketOperator(task_id="create_bucket", bucket_name=bucket_name)
 
-    create_opensearch_policies = create_opensearch_policies(
+    opensearch_policies = create_opensearch_policies(
         bedrock_role_arn=test_context[ROLE_ARN_KEY],
         collection_name=vector_store_name,
         policy_name_suffix=env_id,
@@ -563,7 +563,7 @@ def delete_opensearch_policies(collection_name: str):
         # TEST SETUP
         test_context,
         create_bucket,
-        create_opensearch_policies,
+        opensearch_policies,
         collection,
         await_collection,
         create_vector_index(index_name=index_name, collection_id=collection, region=region_name),