From 6ba2658aef1ed457f42188203fc9948cf5ae54c0 Mon Sep 17 00:00:00 2001
From: Gopal Dirisala <39794726+dirrao@users.noreply.github.com>
Date: Fri, 16 Aug 2024 19:20:23 +0530
Subject: [PATCH] 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                      | 24 +------------------
 newsfragments/41520.significant.rst           |  5 ++++
 .../example_bedrock_retrieve_and_generate.py  |  6 ++---
 3 files changed, 9 insertions(+), 26 deletions(-)
 create mode 100644 newsfragments/41520.significant.rst

diff --git a/airflow/utils/helpers.py b/airflow/utils/helpers.py
index 801cda03f017e..8c08acc66be24 100644
--- a/airflow/utils/helpers.py
+++ b/airflow/utils/helpers.py
@@ -21,7 +21,6 @@
 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
@@ -29,8 +28,7 @@
 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.
diff --git a/newsfragments/41520.significant.rst b/newsfragments/41520.significant.rst
new file mode 100644
index 0000000000000..6f5c605da59a5
--- /dev/null
+++ b/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``)
diff --git a/tests/system/providers/amazon/aws/example_bedrock_retrieve_and_generate.py b/tests/system/providers/amazon/aws/example_bedrock_retrieve_and_generate.py
index fcebc8c40a0d4..224ddc21d8e87 100644
--- a/tests/system/providers/amazon/aws/example_bedrock_retrieve_and_generate.py
+++ b/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),