Remove some useless try/except from providers code (#33967)

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

@@ -21,7 +21,6 @@
 from typing import Any, Sequence
 
 import botocore.exceptions
-from botocore.exceptions import ClientError
 
 from airflow.exceptions import AirflowProviderDeprecationWarning
 from airflow.providers.amazon.aws.hooks.base_aws import AwsBaseAsyncHook, AwsBaseHook
@@ -70,17 +69,14 @@ def create_cluster(
             for the cluster that is being created.
         :param params: Remaining AWS Create cluster API params.
         """
-        try:
-            response = self.get_conn().create_cluster(
-                ClusterIdentifier=cluster_identifier,
-                NodeType=node_type,
-                MasterUsername=master_username,
-                MasterUserPassword=master_user_password,
-                **params,
-            )
-            return response
-        except ClientError as e:
-            raise e
+        response = self.get_conn().create_cluster(
+            ClusterIdentifier=cluster_identifier,
+            NodeType=node_type,
+            MasterUsername=master_username,
+            MasterUserPassword=master_user_password,
+            **params,
+        )
+        return response
 
     # TODO: Wrap create_cluster_snapshot
     def cluster_status(self, cluster_identifier: str) -> str:

airflow/providers/dbt/cloud/hooks/dbt.py

@@ -254,15 +254,10 @@ async def get_job_status(
         :param include_related: Optional. List of related fields to pull with the run.
             Valid values are "trigger", "job", "repository", and "environment".
         """
-        try:
-            self.log.info("Getting the status of job run %s.", run_id)
-            response = await self.get_job_details(
-                run_id, account_id=account_id, include_related=include_related
-            )
-            job_run_status: int = response["data"]["status"]
-            return job_run_status
-        except Exception as e:
-            raise e
+        self.log.info("Getting the status of job run %s.", run_id)
+        response = await self.get_job_details(run_id, account_id=account_id, include_related=include_related)
+        job_run_status: int = response["data"]["status"]
+        return job_run_status
 
     @cached_property
     def connection(self) -> Connection:

airflow/providers/google/cloud/hooks/cloud_sql.py

@@ -437,15 +437,12 @@ async def get_operation_name(self, project_id: str, operation_name: str, session
 
     async def get_operation(self, project_id: str, operation_name: str):
         async with ClientSession() as session:
-            try:
-                operation = await self.get_operation_name(
-                    project_id=project_id,
-                    operation_name=operation_name,
-                    session=session,
-                )
-                operation = await operation.json(content_type=None)
-            except HttpError as e:
-                raise e
+            operation = await self.get_operation_name(
+                project_id=project_id,
+                operation_name=operation_name,
+                session=session,
+            )
+            operation = await operation.json(content_type=None)
             return operation