feat(amazon): add deprecation warning to redshift_cluster triggerers …

…and hooks

astronomer/providers/amazon/aws/hooks/redshift_cluster.py

@@ -1,5 +1,8 @@
+from __future__ import annotations
+
 import asyncio
-from typing import Any, Dict, Optional
+import warnings
+from typing import Any
 
 import botocore.exceptions
 
@@ -10,11 +13,19 @@ class RedshiftHookAsync(AwsBaseHookAsync):
     """Interact with AWS Redshift using aiobotocore python library"""
 
     def __init__(self, *args: Any, **kwargs: Any) -> None:
+        warnings.warn(
+            (
+                "This module is deprecated and will be removed in 2.0.0."
+                "Please use :class: `~airflow.providers.amazon.aws.hooks.redshift_cluster.RedshiftHook`."
+            ),
+            DeprecationWarning,
+            stacklevel=2,
+        )
         kwargs["client_type"] = "redshift"
         kwargs["resource_type"] = "redshift"
         super().__init__(*args, **kwargs)
 
-    async def cluster_status(self, cluster_identifier: str, delete_operation: bool = False) -> Dict[str, Any]:
+    async def cluster_status(self, cluster_identifier: str, delete_operation: bool = False) -> dict[str, Any]:
         """
         Connects to the AWS redshift cluster via aiobotocore and get the status
         and returns the status of the cluster based on the cluster_identifier passed
@@ -38,9 +49,9 @@ async def delete_cluster(
         self,
         cluster_identifier: str,
         skip_final_cluster_snapshot: bool = True,
-        final_cluster_snapshot_identifier: Optional[str] = None,
+        final_cluster_snapshot_identifier: str | None = None,
         polling_period_seconds: float = 5.0,
-    ) -> Dict[str, Any]:
+    ) -> dict[str, Any]:
         """
         Connects to the AWS redshift cluster via aiobotocore and
         deletes the cluster based on the cluster_identifier passed
@@ -77,7 +88,7 @@ async def delete_cluster(
 
     async def pause_cluster(
         self, cluster_identifier: str, polling_period_seconds: float = 5.0
-    ) -> Dict[str, Any]:
+    ) -> dict[str, Any]:
         """
         Connects to the AWS redshift cluster via aiobotocore and
         pause the cluster based on the cluster_identifier passed
@@ -106,7 +117,7 @@ async def resume_cluster(
         self,
         cluster_identifier: str,
         polling_period_seconds: float = 5.0,
-    ) -> Dict[str, Any]:
+    ) -> dict[str, Any]:
         """
         Connects to the AWS redshift cluster via aiobotocore and
         resume the cluster for the cluster_identifier passed
@@ -137,7 +148,7 @@ async def get_cluster_status(
         expected_state: str,
         flag: asyncio.Event,
         delete_operation: bool = False,
-    ) -> Dict[str, Any]:
+    ) -> dict[str, Any]:
         """
         Make call self.cluster_status to know the status and run till the expected_state is met and set the flag
 

astronomer/providers/amazon/aws/operators/redshift_cluster.py

@@ -50,7 +50,7 @@ class RedshiftPauseClusterOperatorAsync(RedshiftPauseClusterOperator):
     This class is deprecated.
     Please use :class: `~airflow.providers.amazon.aws.operators.redshift_cluster.RedshiftPauseClusterOperator`.
     """
-    
+
     def __init__(self, *args, **kwargs):  # type: ignore[no-untyped-def]
         warnings.warn(
             (

astronomer/providers/amazon/aws/triggers/redshift_cluster.py

@@ -1,4 +1,5 @@
 import asyncio
+import warnings
 from typing import Any, AsyncIterator, Dict, Optional, Tuple
 
 from airflow.triggers.base import BaseTrigger, TriggerEvent
@@ -30,6 +31,14 @@ def __init__(
         skip_final_cluster_snapshot: bool = True,
         final_cluster_snapshot_identifier: Optional[str] = None,
     ):
+        warnings.warn(
+            (
+                "This module is deprecated and will be removed in 2.0.0."
+                "Please use hooks in :module: `~airflow.providers.amazon.aws.triggers.redshift_cluster`."
+            ),
+            DeprecationWarning,
+            stacklevel=2,
+        )
         super().__init__()
         self.task_id = task_id
         self.polling_period_seconds = polling_period_seconds