From 667ab8c6eaceb689f6a1afbd909d88bdf0584342 Mon Sep 17 00:00:00 2001
From: Hussein Awala <hussein@awala.fr>
Date: Mon, 28 Aug 2023 22:59:27 +0200
Subject: [PATCH] Improve modules import in AWS probvider by move some of them
 into a type-checking block (#33780)

* Improve modules import in AWS probvider by move some of them into a type-checking block

* comment from code review and fix static checks
---
 airflow/providers/amazon/aws/hooks/athena.py           |  7 ++++---
 airflow/providers/amazon/aws/hooks/base_aws.py         |  5 +++--
 airflow/providers/amazon/aws/hooks/batch_client.py     |  6 ++++--
 airflow/providers/amazon/aws/hooks/batch_waiters.py    |  6 ++++--
 airflow/providers/amazon/aws/hooks/cloud_formation.py  |  6 +++++-
 airflow/providers/amazon/aws/hooks/ecr.py              |  5 ++++-
 airflow/providers/amazon/aws/hooks/ecs.py              |  5 ++++-
 airflow/providers/amazon/aws/links/emr.py              |  7 ++++---
 .../amazon/aws/log/cloudwatch_task_handler.py          |  5 ++++-
 airflow/providers/amazon/aws/notifications/chime.py    |  5 ++++-
 airflow/providers/amazon/aws/operators/ecs.py          |  4 ++--
 airflow/providers/amazon/aws/operators/rds.py          |  4 ++--
 airflow/providers/amazon/aws/sensors/ecs.py            |  4 ++--
 airflow/providers/amazon/aws/sensors/sqs.py            |  2 +-
 airflow/providers/amazon/aws/transfers/mongo_to_s3.py  |  5 +++--
 airflow/providers/amazon/aws/transfers/sql_to_s3.py    |  2 +-
 airflow/providers/amazon/aws/triggers/athena.py        |  6 +++++-
 airflow/providers/amazon/aws/triggers/base.py          |  6 ++++--
 airflow/providers/amazon/aws/triggers/batch.py         |  6 ++++--
 airflow/providers/amazon/aws/triggers/ecs.py           |  6 ++++--
 airflow/providers/amazon/aws/triggers/eks.py           |  6 ++++--
 airflow/providers/amazon/aws/triggers/emr.py           |  6 ++++--
 airflow/providers/amazon/aws/triggers/glue_crawler.py  |  5 ++++-
 .../providers/amazon/aws/triggers/lambda_function.py   |  6 +++++-
 airflow/providers/amazon/aws/triggers/rds.py           |  6 ++++--
 .../providers/amazon/aws/triggers/redshift_cluster.py  |  5 ++++-
 airflow/providers/amazon/aws/triggers/s3.py            |  6 ++++--
 airflow/providers/amazon/aws/triggers/sqs.py           |  6 ++++--
 airflow/providers/amazon/aws/triggers/step_function.py |  6 +++++-
 airflow/providers/amazon/aws/utils/redshift.py         |  4 +++-
 airflow/providers/amazon/aws/utils/task_log_fetcher.py |  6 ++++--
 .../providers/amazon/aws/utils/waiter_with_logging.py  |  6 ++++--
 airflow/providers/amazon/aws/waiters/base_waiter.py    |  6 +++++-
 tests/providers/amazon/aws/hooks/test_eks.py           | 10 +++++++---
 tests/providers/amazon/aws/hooks/test_glue.py          |  5 ++++-
 tests/providers/amazon/aws/hooks/test_glue_crawler.py  |  5 ++++-
 tests/providers/amazon/aws/links/conftest.py           |  6 +++++-
 .../amazon/aws/operators/test_emr_serverless.py        |  5 ++++-
 .../providers/amazon/aws/operators/test_eventbridge.py |  5 ++++-
 tests/providers/amazon/aws/operators/test_glue.py      |  5 ++++-
 tests/providers/amazon/aws/operators/test_rds.py       |  5 ++++-
 .../amazon/aws/operators/test_sagemaker_pipeline.py    |  5 ++++-
 tests/providers/amazon/aws/sensors/test_emr_base.py    |  6 +++++-
 tests/providers/amazon/aws/triggers/test_base.py       |  7 +++++--
 tests/providers/amazon/aws/triggers/test_ecs.py        |  5 ++++-
 tests/providers/amazon/aws/utils/eks_test_utils.py     | 10 ++++++----
 46 files changed, 182 insertions(+), 73 deletions(-)

diff --git a/airflow/providers/amazon/aws/hooks/athena.py b/airflow/providers/amazon/aws/hooks/athena.py
index 27e194d3aa8d0..5aa889e10e8ae 100644
--- a/airflow/providers/amazon/aws/hooks/athena.py
+++ b/airflow/providers/amazon/aws/hooks/athena.py
@@ -25,14 +25,15 @@
 from __future__ import annotations
 
 import warnings
-from typing import Any
-
-from botocore.paginate import PageIterator
+from typing import TYPE_CHECKING, Any
 
 from airflow.exceptions import AirflowException, AirflowProviderDeprecationWarning
 from airflow.providers.amazon.aws.hooks.base_aws import AwsBaseHook
 from airflow.providers.amazon.aws.utils.waiter_with_logging import wait
 
+if TYPE_CHECKING:
+    from botocore.paginate import PageIterator
+
 
 class AthenaHook(AwsBaseHook):
     """Interact with Amazon Athena.
diff --git a/airflow/providers/amazon/aws/hooks/base_aws.py b/airflow/providers/amazon/aws/hooks/base_aws.py
index 660d6fa92e753..e272810fd0f0c 100644
--- a/airflow/providers/amazon/aws/hooks/base_aws.py
+++ b/airflow/providers/amazon/aws/hooks/base_aws.py
@@ -43,9 +43,7 @@
 import jinja2
 import requests
 import tenacity
-from botocore.client import ClientMeta
 from botocore.config import Config
-from botocore.credentials import ReadOnlyCredentials
 from botocore.waiter import Waiter, WaiterModel
 from dateutil.tz import tzlocal
 from slugify import slugify
@@ -66,6 +64,9 @@
 BaseAwsConnection = TypeVar("BaseAwsConnection", bound=Union[boto3.client, boto3.resource])
 
 if TYPE_CHECKING:
+    from botocore.client import ClientMeta
+    from botocore.credentials import ReadOnlyCredentials
+
     from airflow.models.connection import Connection  # Avoid circular imports.
 
 
diff --git a/airflow/providers/amazon/aws/hooks/batch_client.py b/airflow/providers/amazon/aws/hooks/batch_client.py
index 26304ed3673e9..6eda1d29667a3 100644
--- a/airflow/providers/amazon/aws/hooks/batch_client.py
+++ b/airflow/providers/amazon/aws/hooks/batch_client.py
@@ -29,7 +29,7 @@
 import itertools
 from random import uniform
 from time import sleep
-from typing import Callable
+from typing import TYPE_CHECKING, Callable
 
 import botocore.client
 import botocore.exceptions
@@ -37,9 +37,11 @@
 
 from airflow.exceptions import AirflowException
 from airflow.providers.amazon.aws.hooks.base_aws import AwsBaseHook
-from airflow.providers.amazon.aws.utils.task_log_fetcher import AwsTaskLogFetcher
 from airflow.typing_compat import Protocol, runtime_checkable
 
+if TYPE_CHECKING:
+    from airflow.providers.amazon.aws.utils.task_log_fetcher import AwsTaskLogFetcher
+
 
 @runtime_checkable
 class BatchProtocol(Protocol):
diff --git a/airflow/providers/amazon/aws/hooks/batch_waiters.py b/airflow/providers/amazon/aws/hooks/batch_waiters.py
index c93a00b0537d4..afd9c3bc599da 100644
--- a/airflow/providers/amazon/aws/hooks/batch_waiters.py
+++ b/airflow/providers/amazon/aws/hooks/batch_waiters.py
@@ -29,7 +29,7 @@
 import sys
 from copy import deepcopy
 from pathlib import Path
-from typing import Callable
+from typing import TYPE_CHECKING, Callable
 
 import botocore.client
 import botocore.exceptions
@@ -37,7 +37,9 @@
 
 from airflow.exceptions import AirflowException
 from airflow.providers.amazon.aws.hooks.batch_client import BatchClientHook
-from airflow.providers.amazon.aws.utils.task_log_fetcher import AwsTaskLogFetcher
+
+if TYPE_CHECKING:
+    from airflow.providers.amazon.aws.utils.task_log_fetcher import AwsTaskLogFetcher
 
 
 class BatchWaitersHook(BatchClientHook):
diff --git a/airflow/providers/amazon/aws/hooks/cloud_formation.py b/airflow/providers/amazon/aws/hooks/cloud_formation.py
index 1772e6501c1ac..5250b3c430b9e 100644
--- a/airflow/providers/amazon/aws/hooks/cloud_formation.py
+++ b/airflow/providers/amazon/aws/hooks/cloud_formation.py
@@ -18,11 +18,15 @@
 """This module contains AWS CloudFormation Hook."""
 from __future__ import annotations
 
-from boto3 import client, resource
+from typing import TYPE_CHECKING
+
 from botocore.exceptions import ClientError
 
 from airflow.providers.amazon.aws.hooks.base_aws import AwsBaseHook
 
+if TYPE_CHECKING:
+    from boto3 import client, resource
+
 
 class CloudFormationHook(AwsBaseHook):
     """
diff --git a/airflow/providers/amazon/aws/hooks/ecr.py b/airflow/providers/amazon/aws/hooks/ecr.py
index f60d4d070d067..7ee725a16e5a8 100644
--- a/airflow/providers/amazon/aws/hooks/ecr.py
+++ b/airflow/providers/amazon/aws/hooks/ecr.py
@@ -20,11 +20,14 @@
 import base64
 import logging
 from dataclasses import dataclass
-from datetime import datetime
+from typing import TYPE_CHECKING
 
 from airflow.providers.amazon.aws.hooks.base_aws import AwsBaseHook
 from airflow.utils.log.secrets_masker import mask_secret
 
+if TYPE_CHECKING:
+    from datetime import datetime
+
 logger = logging.getLogger(__name__)
 
 
diff --git a/airflow/providers/amazon/aws/hooks/ecs.py b/airflow/providers/amazon/aws/hooks/ecs.py
index 6d6f8cd82114c..8762655386c66 100644
--- a/airflow/providers/amazon/aws/hooks/ecs.py
+++ b/airflow/providers/amazon/aws/hooks/ecs.py
@@ -17,13 +17,16 @@
 # under the License.
 from __future__ import annotations
 
-from botocore.waiter import Waiter
+from typing import TYPE_CHECKING
 
 from airflow.providers.amazon.aws.exceptions import EcsOperatorError, EcsTaskFailToStart
 from airflow.providers.amazon.aws.hooks.base_aws import AwsGenericHook
 from airflow.providers.amazon.aws.utils import _StringCompareEnum
 from airflow.typing_compat import Protocol, runtime_checkable
 
+if TYPE_CHECKING:
+    from botocore.waiter import Waiter
+
 
 def should_retry(exception: Exception):
     """Check if exception is related to ECS resource quota (CPU, MEM)."""
diff --git a/airflow/providers/amazon/aws/links/emr.py b/airflow/providers/amazon/aws/links/emr.py
index cacb4ff5fb63e..6c8cd2181eee1 100644
--- a/airflow/providers/amazon/aws/links/emr.py
+++ b/airflow/providers/amazon/aws/links/emr.py
@@ -16,15 +16,16 @@
 # under the License.
 from __future__ import annotations
 
-from typing import Any
-
-import boto3
+from typing import TYPE_CHECKING, Any
 
 from airflow.exceptions import AirflowException
 from airflow.providers.amazon.aws.hooks.s3 import S3Hook
 from airflow.providers.amazon.aws.links.base_aws import BASE_AWS_CONSOLE_LINK, BaseAwsLink
 from airflow.utils.helpers import exactly_one
 
+if TYPE_CHECKING:
+    import boto3
+
 
 class EmrClusterLink(BaseAwsLink):
     """Helper class for constructing AWS EMR Cluster Link."""
diff --git a/airflow/providers/amazon/aws/log/cloudwatch_task_handler.py b/airflow/providers/amazon/aws/log/cloudwatch_task_handler.py
index 374eb29b2f605..4d3ccec00c6e6 100644
--- a/airflow/providers/amazon/aws/log/cloudwatch_task_handler.py
+++ b/airflow/providers/amazon/aws/log/cloudwatch_task_handler.py
@@ -19,16 +19,19 @@
 
 from datetime import datetime, timedelta
 from functools import cached_property
+from typing import TYPE_CHECKING
 
 import watchtower
 
 from airflow.configuration import conf
-from airflow.models import TaskInstance
 from airflow.providers.amazon.aws.hooks.logs import AwsLogsHook
 from airflow.providers.amazon.aws.utils import datetime_to_epoch_utc_ms
 from airflow.utils.log.file_task_handler import FileTaskHandler
 from airflow.utils.log.logging_mixin import LoggingMixin
 
+if TYPE_CHECKING:
+    from airflow.models import TaskInstance
+
 
 class CloudwatchTaskHandler(FileTaskHandler, LoggingMixin):
     """
diff --git a/airflow/providers/amazon/aws/notifications/chime.py b/airflow/providers/amazon/aws/notifications/chime.py
index d62b4b21d6fa3..801666f7729ad 100644
--- a/airflow/providers/amazon/aws/notifications/chime.py
+++ b/airflow/providers/amazon/aws/notifications/chime.py
@@ -18,10 +18,13 @@
 from __future__ import annotations
 
 from functools import cached_property
+from typing import TYPE_CHECKING
 
 from airflow.exceptions import AirflowOptionalProviderFeatureException
 from airflow.providers.amazon.aws.hooks.chime import ChimeWebhookHook
-from airflow.utils.context import Context
+
+if TYPE_CHECKING:
+    from airflow.utils.context import Context
 
 try:
     from airflow.notifications.basenotifier import BaseNotifier
diff --git a/airflow/providers/amazon/aws/operators/ecs.py b/airflow/providers/amazon/aws/operators/ecs.py
index e5d5e7d1b0ba0..ed8e5ccec35f8 100644
--- a/airflow/providers/amazon/aws/operators/ecs.py
+++ b/airflow/providers/amazon/aws/operators/ecs.py
@@ -24,8 +24,6 @@
 from functools import cached_property
 from typing import TYPE_CHECKING, Sequence
 
-import boto3
-
 from airflow.configuration import conf
 from airflow.exceptions import AirflowException, AirflowProviderDeprecationWarning
 from airflow.models import BaseOperator
@@ -43,6 +41,8 @@
 from airflow.utils.helpers import prune_dict
 
 if TYPE_CHECKING:
+    import boto3
+
     from airflow.models import TaskInstance
     from airflow.utils.context import Context
 
diff --git a/airflow/providers/amazon/aws/operators/rds.py b/airflow/providers/amazon/aws/operators/rds.py
index f7361892b059c..6d1f2fd7fb4e5 100644
--- a/airflow/providers/amazon/aws/operators/rds.py
+++ b/airflow/providers/amazon/aws/operators/rds.py
@@ -23,8 +23,6 @@
 from functools import cached_property
 from typing import TYPE_CHECKING, Any, Sequence
 
-from mypy_boto3_rds.type_defs import TagTypeDef
-
 from airflow.configuration import conf
 from airflow.exceptions import AirflowException, AirflowProviderDeprecationWarning
 from airflow.models import BaseOperator
@@ -39,6 +37,8 @@
 from airflow.providers.amazon.aws.utils.waiter_with_logging import wait
 
 if TYPE_CHECKING:
+    from mypy_boto3_rds.type_defs import TagTypeDef
+
     from airflow.utils.context import Context
 
 
diff --git a/airflow/providers/amazon/aws/sensors/ecs.py b/airflow/providers/amazon/aws/sensors/ecs.py
index 8580d0c4795d2..320ef84b834b5 100644
--- a/airflow/providers/amazon/aws/sensors/ecs.py
+++ b/airflow/providers/amazon/aws/sensors/ecs.py
@@ -19,8 +19,6 @@
 from functools import cached_property
 from typing import TYPE_CHECKING, Sequence
 
-import boto3
-
 from airflow.exceptions import AirflowException
 from airflow.providers.amazon.aws.hooks.ecs import (
     EcsClusterStates,
@@ -31,6 +29,8 @@
 from airflow.sensors.base import BaseSensorOperator
 
 if TYPE_CHECKING:
+    import boto3
+
     from airflow.utils.context import Context
 
 DEFAULT_CONN_ID: str = "aws_default"
diff --git a/airflow/providers/amazon/aws/sensors/sqs.py b/airflow/providers/amazon/aws/sensors/sqs.py
index 43191d1445e10..aca950375a282 100644
--- a/airflow/providers/amazon/aws/sensors/sqs.py
+++ b/airflow/providers/amazon/aws/sensors/sqs.py
@@ -26,13 +26,13 @@
 
 from airflow.configuration import conf
 from airflow.exceptions import AirflowException
-from airflow.providers.amazon.aws.hooks.base_aws import BaseAwsConnection
 from airflow.providers.amazon.aws.hooks.sqs import SqsHook
 from airflow.providers.amazon.aws.triggers.sqs import SqsSensorTrigger
 from airflow.providers.amazon.aws.utils.sqs import process_response
 from airflow.sensors.base import BaseSensorOperator
 
 if TYPE_CHECKING:
+    from airflow.providers.amazon.aws.hooks.base_aws import BaseAwsConnection
     from airflow.utils.context import Context
 from datetime import timedelta
 
diff --git a/airflow/providers/amazon/aws/transfers/mongo_to_s3.py b/airflow/providers/amazon/aws/transfers/mongo_to_s3.py
index dcfd0d6c700ac..d7a0eae150adf 100644
--- a/airflow/providers/amazon/aws/transfers/mongo_to_s3.py
+++ b/airflow/providers/amazon/aws/transfers/mongo_to_s3.py
@@ -21,14 +21,15 @@
 from typing import TYPE_CHECKING, Any, Iterable, Sequence, cast
 
 from bson import json_util
-from pymongo.command_cursor import CommandCursor
-from pymongo.cursor import Cursor
 
 from airflow.models import BaseOperator
 from airflow.providers.amazon.aws.hooks.s3 import S3Hook
 from airflow.providers.mongo.hooks.mongo import MongoHook
 
 if TYPE_CHECKING:
+    from pymongo.command_cursor import CommandCursor
+    from pymongo.cursor import Cursor
+
     from airflow.utils.context import Context
 
 
diff --git a/airflow/providers/amazon/aws/transfers/sql_to_s3.py b/airflow/providers/amazon/aws/transfers/sql_to_s3.py
index e0860620234bc..79f5b65e3ef08 100644
--- a/airflow/providers/amazon/aws/transfers/sql_to_s3.py
+++ b/airflow/providers/amazon/aws/transfers/sql_to_s3.py
@@ -28,11 +28,11 @@
 from airflow.hooks.base import BaseHook
 from airflow.models import BaseOperator
 from airflow.providers.amazon.aws.hooks.s3 import S3Hook
-from airflow.providers.common.sql.hooks.sql import DbApiHook
 
 if TYPE_CHECKING:
     import pandas as pd
 
+    from airflow.providers.common.sql.hooks.sql import DbApiHook
     from airflow.utils.context import Context
 
 
diff --git a/airflow/providers/amazon/aws/triggers/athena.py b/airflow/providers/amazon/aws/triggers/athena.py
index fcd09dae19039..65c4f08000c0f 100644
--- a/airflow/providers/amazon/aws/triggers/athena.py
+++ b/airflow/providers/amazon/aws/triggers/athena.py
@@ -16,10 +16,14 @@
 # under the License.
 from __future__ import annotations
 
+from typing import TYPE_CHECKING
+
 from airflow.providers.amazon.aws.hooks.athena import AthenaHook
-from airflow.providers.amazon.aws.hooks.base_aws import AwsGenericHook
 from airflow.providers.amazon.aws.triggers.base import AwsBaseWaiterTrigger
 
+if TYPE_CHECKING:
+    from airflow.providers.amazon.aws.hooks.base_aws import AwsGenericHook
+
 
 class AthenaTrigger(AwsBaseWaiterTrigger):
     """
diff --git a/airflow/providers/amazon/aws/triggers/base.py b/airflow/providers/amazon/aws/triggers/base.py
index d2d664d97f521..a6fc6104dd286 100644
--- a/airflow/providers/amazon/aws/triggers/base.py
+++ b/airflow/providers/amazon/aws/triggers/base.py
@@ -18,12 +18,14 @@
 from __future__ import annotations
 
 from abc import abstractmethod
-from typing import Any, AsyncIterator
+from typing import TYPE_CHECKING, Any, AsyncIterator
 
-from airflow.providers.amazon.aws.hooks.base_aws import AwsGenericHook
 from airflow.providers.amazon.aws.utils.waiter_with_logging import async_wait
 from airflow.triggers.base import BaseTrigger, TriggerEvent
 
+if TYPE_CHECKING:
+    from airflow.providers.amazon.aws.hooks.base_aws import AwsGenericHook
+
 
 class AwsBaseWaiterTrigger(BaseTrigger):
     """
diff --git a/airflow/providers/amazon/aws/triggers/batch.py b/airflow/providers/amazon/aws/triggers/batch.py
index 900040afa231a..2bf9e950620c8 100644
--- a/airflow/providers/amazon/aws/triggers/batch.py
+++ b/airflow/providers/amazon/aws/triggers/batch.py
@@ -19,16 +19,18 @@
 import asyncio
 import itertools
 from functools import cached_property
-from typing import Any
+from typing import TYPE_CHECKING, Any
 
 from botocore.exceptions import WaiterError
 from deprecated import deprecated
 
-from airflow.providers.amazon.aws.hooks.base_aws import AwsGenericHook
 from airflow.providers.amazon.aws.hooks.batch_client import BatchClientHook
 from airflow.providers.amazon.aws.triggers.base import AwsBaseWaiterTrigger
 from airflow.triggers.base import BaseTrigger, TriggerEvent
 
+if TYPE_CHECKING:
+    from airflow.providers.amazon.aws.hooks.base_aws import AwsGenericHook
+
 
 @deprecated(reason="use BatchJobTrigger instead")
 class BatchOperatorTrigger(BaseTrigger):
diff --git a/airflow/providers/amazon/aws/triggers/ecs.py b/airflow/providers/amazon/aws/triggers/ecs.py
index ea183e183d88a..e0c8212940095 100644
--- a/airflow/providers/amazon/aws/triggers/ecs.py
+++ b/airflow/providers/amazon/aws/triggers/ecs.py
@@ -18,18 +18,20 @@
 from __future__ import annotations
 
 import asyncio
-from typing import Any, AsyncIterator
+from typing import TYPE_CHECKING, Any, AsyncIterator
 
 from botocore.exceptions import ClientError, WaiterError
 
 from airflow import AirflowException
-from airflow.providers.amazon.aws.hooks.base_aws import AwsGenericHook
 from airflow.providers.amazon.aws.hooks.ecs import EcsHook
 from airflow.providers.amazon.aws.hooks.logs import AwsLogsHook
 from airflow.providers.amazon.aws.triggers.base import AwsBaseWaiterTrigger
 from airflow.providers.amazon.aws.utils.task_log_fetcher import AwsTaskLogFetcher
 from airflow.triggers.base import BaseTrigger, TriggerEvent
 
+if TYPE_CHECKING:
+    from airflow.providers.amazon.aws.hooks.base_aws import AwsGenericHook
+
 
 class ClusterActiveTrigger(AwsBaseWaiterTrigger):
     """
diff --git a/airflow/providers/amazon/aws/triggers/eks.py b/airflow/providers/amazon/aws/triggers/eks.py
index 01f66d4567005..5a48ee39c0758 100644
--- a/airflow/providers/amazon/aws/triggers/eks.py
+++ b/airflow/providers/amazon/aws/triggers/eks.py
@@ -17,15 +17,17 @@
 from __future__ import annotations
 
 import warnings
-from typing import Any
+from typing import TYPE_CHECKING, Any
 
 from airflow.exceptions import AirflowProviderDeprecationWarning
-from airflow.providers.amazon.aws.hooks.base_aws import AwsGenericHook
 from airflow.providers.amazon.aws.hooks.eks import EksHook
 from airflow.providers.amazon.aws.triggers.base import AwsBaseWaiterTrigger
 from airflow.providers.amazon.aws.utils.waiter_with_logging import async_wait
 from airflow.triggers.base import TriggerEvent
 
+if TYPE_CHECKING:
+    from airflow.providers.amazon.aws.hooks.base_aws import AwsGenericHook
+
 
 class EksCreateClusterTrigger(AwsBaseWaiterTrigger):
     """
diff --git a/airflow/providers/amazon/aws/triggers/emr.py b/airflow/providers/amazon/aws/triggers/emr.py
index af8fef5f7bcf3..a69fde2a6e52d 100644
--- a/airflow/providers/amazon/aws/triggers/emr.py
+++ b/airflow/providers/amazon/aws/triggers/emr.py
@@ -18,16 +18,18 @@
 
 import asyncio
 import warnings
-from typing import Any
+from typing import TYPE_CHECKING, Any
 
 from botocore.exceptions import WaiterError
 
 from airflow.exceptions import AirflowProviderDeprecationWarning
-from airflow.providers.amazon.aws.hooks.base_aws import AwsGenericHook
 from airflow.providers.amazon.aws.hooks.emr import EmrContainerHook, EmrHook, EmrServerlessHook
 from airflow.providers.amazon.aws.triggers.base import AwsBaseWaiterTrigger
 from airflow.triggers.base import BaseTrigger, TriggerEvent
 
+if TYPE_CHECKING:
+    from airflow.providers.amazon.aws.hooks.base_aws import AwsGenericHook
+
 
 class EmrAddStepsTrigger(BaseTrigger):
     """
diff --git a/airflow/providers/amazon/aws/triggers/glue_crawler.py b/airflow/providers/amazon/aws/triggers/glue_crawler.py
index 092b05242983a..78aa295838051 100644
--- a/airflow/providers/amazon/aws/triggers/glue_crawler.py
+++ b/airflow/providers/amazon/aws/triggers/glue_crawler.py
@@ -17,12 +17,15 @@
 from __future__ import annotations
 
 import warnings
+from typing import TYPE_CHECKING
 
 from airflow.exceptions import AirflowProviderDeprecationWarning
-from airflow.providers.amazon.aws.hooks.base_aws import AwsGenericHook
 from airflow.providers.amazon.aws.hooks.glue_crawler import GlueCrawlerHook
 from airflow.providers.amazon.aws.triggers.base import AwsBaseWaiterTrigger
 
+if TYPE_CHECKING:
+    from airflow.providers.amazon.aws.hooks.base_aws import AwsGenericHook
+
 
 class GlueCrawlerCompleteTrigger(AwsBaseWaiterTrigger):
     """
diff --git a/airflow/providers/amazon/aws/triggers/lambda_function.py b/airflow/providers/amazon/aws/triggers/lambda_function.py
index f0f6a40551b96..4805d5bb6cf3e 100644
--- a/airflow/providers/amazon/aws/triggers/lambda_function.py
+++ b/airflow/providers/amazon/aws/triggers/lambda_function.py
@@ -16,10 +16,14 @@
 # under the License.
 from __future__ import annotations
 
-from airflow.providers.amazon.aws.hooks.base_aws import AwsGenericHook
+from typing import TYPE_CHECKING
+
 from airflow.providers.amazon.aws.hooks.lambda_function import LambdaHook
 from airflow.providers.amazon.aws.triggers.base import AwsBaseWaiterTrigger
 
+if TYPE_CHECKING:
+    from airflow.providers.amazon.aws.hooks.base_aws import AwsGenericHook
+
 
 class LambdaCreateFunctionCompleteTrigger(AwsBaseWaiterTrigger):
     """
diff --git a/airflow/providers/amazon/aws/triggers/rds.py b/airflow/providers/amazon/aws/triggers/rds.py
index 82354c1874112..ebc80ba700a2c 100644
--- a/airflow/providers/amazon/aws/triggers/rds.py
+++ b/airflow/providers/amazon/aws/triggers/rds.py
@@ -17,16 +17,18 @@
 from __future__ import annotations
 
 import warnings
-from typing import Any
+from typing import TYPE_CHECKING, Any
 
 from airflow.exceptions import AirflowProviderDeprecationWarning
-from airflow.providers.amazon.aws.hooks.base_aws import AwsGenericHook
 from airflow.providers.amazon.aws.hooks.rds import RdsHook
 from airflow.providers.amazon.aws.triggers.base import AwsBaseWaiterTrigger
 from airflow.providers.amazon.aws.utils.rds import RdsDbType
 from airflow.providers.amazon.aws.utils.waiter_with_logging import async_wait
 from airflow.triggers.base import BaseTrigger, TriggerEvent
 
+if TYPE_CHECKING:
+    from airflow.providers.amazon.aws.hooks.base_aws import AwsGenericHook
+
 
 class RdsDbInstanceTrigger(BaseTrigger):
     """
diff --git a/airflow/providers/amazon/aws/triggers/redshift_cluster.py b/airflow/providers/amazon/aws/triggers/redshift_cluster.py
index 678f435364c08..64a636c8dbcee 100644
--- a/airflow/providers/amazon/aws/triggers/redshift_cluster.py
+++ b/airflow/providers/amazon/aws/triggers/redshift_cluster.py
@@ -17,12 +17,15 @@
 from __future__ import annotations
 
 import warnings
+from typing import TYPE_CHECKING
 
 from airflow.exceptions import AirflowProviderDeprecationWarning
-from airflow.providers.amazon.aws.hooks.base_aws import AwsGenericHook
 from airflow.providers.amazon.aws.hooks.redshift_cluster import RedshiftHook
 from airflow.providers.amazon.aws.triggers.base import AwsBaseWaiterTrigger
 
+if TYPE_CHECKING:
+    from airflow.providers.amazon.aws.hooks.base_aws import AwsGenericHook
+
 
 class RedshiftCreateClusterTrigger(AwsBaseWaiterTrigger):
     """
diff --git a/airflow/providers/amazon/aws/triggers/s3.py b/airflow/providers/amazon/aws/triggers/s3.py
index 06432108eb35f..302abe91d626f 100644
--- a/airflow/providers/amazon/aws/triggers/s3.py
+++ b/airflow/providers/amazon/aws/triggers/s3.py
@@ -17,13 +17,15 @@
 from __future__ import annotations
 
 import asyncio
-from datetime import datetime
 from functools import cached_property
-from typing import Any, AsyncIterator
+from typing import TYPE_CHECKING, Any, AsyncIterator
 
 from airflow.providers.amazon.aws.hooks.s3 import S3Hook
 from airflow.triggers.base import BaseTrigger, TriggerEvent
 
+if TYPE_CHECKING:
+    from datetime import datetime
+
 
 class S3KeyTrigger(BaseTrigger):
     """
diff --git a/airflow/providers/amazon/aws/triggers/sqs.py b/airflow/providers/amazon/aws/triggers/sqs.py
index 7b3b0f312b34c..23bbb44a1828d 100644
--- a/airflow/providers/amazon/aws/triggers/sqs.py
+++ b/airflow/providers/amazon/aws/triggers/sqs.py
@@ -17,16 +17,18 @@
 from __future__ import annotations
 
 import asyncio
-from typing import Any, AsyncIterator, Collection
+from typing import TYPE_CHECKING, Any, AsyncIterator, Collection
 
 from typing_extensions import Literal
 
 from airflow.exceptions import AirflowException
-from airflow.providers.amazon.aws.hooks.base_aws import BaseAwsConnection
 from airflow.providers.amazon.aws.hooks.sqs import SqsHook
 from airflow.providers.amazon.aws.utils.sqs import process_response
 from airflow.triggers.base import BaseTrigger, TriggerEvent
 
+if TYPE_CHECKING:
+    from airflow.providers.amazon.aws.hooks.base_aws import BaseAwsConnection
+
 
 class SqsSensorTrigger(BaseTrigger):
     """
diff --git a/airflow/providers/amazon/aws/triggers/step_function.py b/airflow/providers/amazon/aws/triggers/step_function.py
index c4875f078f196..650a1405d1c93 100644
--- a/airflow/providers/amazon/aws/triggers/step_function.py
+++ b/airflow/providers/amazon/aws/triggers/step_function.py
@@ -16,10 +16,14 @@
 # under the License.
 from __future__ import annotations
 
-from airflow.providers.amazon.aws.hooks.base_aws import AwsGenericHook
+from typing import TYPE_CHECKING
+
 from airflow.providers.amazon.aws.hooks.step_function import StepFunctionHook
 from airflow.providers.amazon.aws.triggers.base import AwsBaseWaiterTrigger
 
+if TYPE_CHECKING:
+    from airflow.providers.amazon.aws.hooks.base_aws import AwsGenericHook
+
 
 class StepFunctionsExecutionCompleteTrigger(AwsBaseWaiterTrigger):
     """
diff --git a/airflow/providers/amazon/aws/utils/redshift.py b/airflow/providers/amazon/aws/utils/redshift.py
index d91858718fecf..2c8d9f6f54abd 100644
--- a/airflow/providers/amazon/aws/utils/redshift.py
+++ b/airflow/providers/amazon/aws/utils/redshift.py
@@ -17,8 +17,10 @@
 from __future__ import annotations
 
 import logging
+from typing import TYPE_CHECKING
 
-from botocore.credentials import ReadOnlyCredentials
+if TYPE_CHECKING:
+    from botocore.credentials import ReadOnlyCredentials
 
 log = logging.getLogger(__name__)
 
diff --git a/airflow/providers/amazon/aws/utils/task_log_fetcher.py b/airflow/providers/amazon/aws/utils/task_log_fetcher.py
index e5cac2f426326..63110edadad96 100644
--- a/airflow/providers/amazon/aws/utils/task_log_fetcher.py
+++ b/airflow/providers/amazon/aws/utils/task_log_fetcher.py
@@ -19,14 +19,16 @@
 
 import time
 from datetime import datetime, timedelta
-from logging import Logger
 from threading import Event, Thread
-from typing import Generator
+from typing import TYPE_CHECKING, Generator
 
 from botocore.exceptions import ClientError, ConnectionClosedError
 
 from airflow.providers.amazon.aws.hooks.logs import AwsLogsHook
 
+if TYPE_CHECKING:
+    from logging import Logger
+
 
 class AwsTaskLogFetcher(Thread):
     """Fetch Cloudwatch log events with specific interval and send the log events to the logger.info."""
diff --git a/airflow/providers/amazon/aws/utils/waiter_with_logging.py b/airflow/providers/amazon/aws/utils/waiter_with_logging.py
index 2c0d403426530..e1192bcbfb866 100644
--- a/airflow/providers/amazon/aws/utils/waiter_with_logging.py
+++ b/airflow/providers/amazon/aws/utils/waiter_with_logging.py
@@ -20,14 +20,16 @@
 import asyncio
 import logging
 import time
-from typing import Any
+from typing import TYPE_CHECKING, Any
 
 import jmespath
 from botocore.exceptions import WaiterError
-from botocore.waiter import Waiter
 
 from airflow.exceptions import AirflowException
 
+if TYPE_CHECKING:
+    from botocore.waiter import Waiter
+
 
 def wait(
     waiter: Waiter,
diff --git a/airflow/providers/amazon/aws/waiters/base_waiter.py b/airflow/providers/amazon/aws/waiters/base_waiter.py
index 488767a084a21..204a591c3fbfe 100644
--- a/airflow/providers/amazon/aws/waiters/base_waiter.py
+++ b/airflow/providers/amazon/aws/waiters/base_waiter.py
@@ -17,9 +17,13 @@
 
 from __future__ import annotations
 
-import boto3
+from typing import TYPE_CHECKING
+
 from botocore.waiter import Waiter, WaiterModel, create_waiter_with_client
 
+if TYPE_CHECKING:
+    import boto3
+
 
 class BaseBotoWaiter:
     """
diff --git a/tests/providers/amazon/aws/hooks/test_eks.py b/tests/providers/amazon/aws/hooks/test_eks.py
index ae9c6c59d555d..6b841e78c6549 100644
--- a/tests/providers/amazon/aws/hooks/test_eks.py
+++ b/tests/providers/amazon/aws/hooks/test_eks.py
@@ -19,19 +19,17 @@
 
 import sys
 from copy import deepcopy
-from datetime import datetime
 from pathlib import Path
+from typing import TYPE_CHECKING
 from unittest import mock
 from urllib.parse import ParseResult, urlsplit
 
 import pytest
 import time_machine
 import yaml
-from _pytest._code import ExceptionInfo
 from botocore.exceptions import ClientError
 from moto import mock_eks
 from moto.core import DEFAULT_ACCOUNT_ID
-from moto.core.exceptions import AWSError
 from moto.eks.exceptions import (
     InvalidParameterException,
     InvalidRequestException,
@@ -95,6 +93,12 @@
     region_matches_partition,
 )
 
+if TYPE_CHECKING:
+    from datetime import datetime
+
+    from moto.core.exceptions import AWSError
+    from pytest import ExceptionInfo
+
 
 @pytest.fixture(scope="function")
 def cluster_builder():
diff --git a/tests/providers/amazon/aws/hooks/test_glue.py b/tests/providers/amazon/aws/hooks/test_glue.py
index 1fae16e339558..d43ef2b8d9ec6 100644
--- a/tests/providers/amazon/aws/hooks/test_glue.py
+++ b/tests/providers/amazon/aws/hooks/test_glue.py
@@ -18,8 +18,8 @@
 from __future__ import annotations
 
 import json
+from typing import TYPE_CHECKING
 from unittest import mock
-from unittest.mock import MagicMock
 
 import boto3
 import pytest
@@ -30,6 +30,9 @@
 from airflow.providers.amazon.aws.hooks.base_aws import AwsBaseHook
 from airflow.providers.amazon.aws.hooks.glue import GlueJobHook
 
+if TYPE_CHECKING:
+    from unittest.mock import MagicMock
+
 
 class TestGlueJobHook:
     def setup_method(self):
diff --git a/tests/providers/amazon/aws/hooks/test_glue_crawler.py b/tests/providers/amazon/aws/hooks/test_glue_crawler.py
index e283812fc08e4..167ffd0d3d166 100644
--- a/tests/providers/amazon/aws/hooks/test_glue_crawler.py
+++ b/tests/providers/amazon/aws/hooks/test_glue_crawler.py
@@ -18,14 +18,17 @@
 from __future__ import annotations
 
 from copy import deepcopy
+from typing import TYPE_CHECKING
 from unittest import mock
-from unittest.mock import MagicMock
 
 from moto import mock_sts
 from moto.core import DEFAULT_ACCOUNT_ID
 
 from airflow.providers.amazon.aws.hooks.glue_crawler import GlueCrawlerHook
 
+if TYPE_CHECKING:
+    from unittest.mock import MagicMock
+
 mock_crawler_name = "test-crawler"
 mock_role_name = "test-role"
 mock_config = {
diff --git a/tests/providers/amazon/aws/links/conftest.py b/tests/providers/amazon/aws/links/conftest.py
index 3d79b327df621..b2b9a4188c993 100644
--- a/tests/providers/amazon/aws/links/conftest.py
+++ b/tests/providers/amazon/aws/links/conftest.py
@@ -16,11 +16,15 @@
 # under the License.
 from __future__ import annotations
 
+from typing import TYPE_CHECKING
+
 import pytest
 
-from airflow.providers.amazon.aws.links.base_aws import BaseAwsLink
 from tests.providers.amazon.aws.utils.links_test_utils import link_test_operator
 
+if TYPE_CHECKING:
+    from airflow.providers.amazon.aws.links.base_aws import BaseAwsLink
+
 
 @pytest.fixture()
 def create_task_and_ti_of_op_with_extra_link(create_task_instance_of_operator):
diff --git a/tests/providers/amazon/aws/operators/test_emr_serverless.py b/tests/providers/amazon/aws/operators/test_emr_serverless.py
index 5e16313d9b52f..a93791e1a1e3f 100644
--- a/tests/providers/amazon/aws/operators/test_emr_serverless.py
+++ b/tests/providers/amazon/aws/operators/test_emr_serverless.py
@@ -16,8 +16,8 @@
 # under the License.
 from __future__ import annotations
 
+from typing import TYPE_CHECKING
 from unittest import mock
-from unittest.mock import MagicMock
 from uuid import UUID
 
 import pytest
@@ -33,6 +33,9 @@
 )
 from airflow.utils.types import NOTSET
 
+if TYPE_CHECKING:
+    from unittest.mock import MagicMock
+
 task_id = "test_emr_serverless_task_id"
 application_id = "test_application_id"
 release_label = "test"
diff --git a/tests/providers/amazon/aws/operators/test_eventbridge.py b/tests/providers/amazon/aws/operators/test_eventbridge.py
index 8ff5703d4f7b4..8d398b22616ae 100644
--- a/tests/providers/amazon/aws/operators/test_eventbridge.py
+++ b/tests/providers/amazon/aws/operators/test_eventbridge.py
@@ -16,8 +16,8 @@
 # under the License.
 from __future__ import annotations
 
+from typing import TYPE_CHECKING
 from unittest import mock
-from unittest.mock import MagicMock
 
 import pytest
 
@@ -30,6 +30,9 @@
     EventBridgePutRuleOperator,
 )
 
+if TYPE_CHECKING:
+    from unittest.mock import MagicMock
+
 ENTRIES = [{"Detail": "test-detail", "Source": "test-source", "DetailType": "test-detail-type"}]
 FAILED_ENTRIES_RESPONSE = [{"ErrorCode": "test_code"}, {"ErrorCode": "test_code"}]
 EVENT_PATTERN = '{"source": ["aws.s3"]}'
diff --git a/tests/providers/amazon/aws/operators/test_glue.py b/tests/providers/amazon/aws/operators/test_glue.py
index dc298563ae65c..ee682f09e07c6 100644
--- a/tests/providers/amazon/aws/operators/test_glue.py
+++ b/tests/providers/amazon/aws/operators/test_glue.py
@@ -16,18 +16,21 @@
 # under the License.
 from __future__ import annotations
 
+from typing import TYPE_CHECKING
 from unittest import mock
 
 import pytest
 
 from airflow.configuration import conf
 from airflow.exceptions import TaskDeferred
-from airflow.models import TaskInstance
 from airflow.providers.amazon.aws.hooks.glue import GlueJobHook
 from airflow.providers.amazon.aws.hooks.s3 import S3Hook
 from airflow.providers.amazon.aws.links.glue import GlueJobRunDetailsLink
 from airflow.providers.amazon.aws.operators.glue import GlueJobOperator
 
+if TYPE_CHECKING:
+    from airflow.models import TaskInstance
+
 TASK_ID = "test_glue_operator"
 DAG_ID = "test_dag_id"
 JOB_NAME = "test_job_name/with_slash"
diff --git a/tests/providers/amazon/aws/operators/test_rds.py b/tests/providers/amazon/aws/operators/test_rds.py
index 47122d55c0ae7..ed1224851c122 100644
--- a/tests/providers/amazon/aws/operators/test_rds.py
+++ b/tests/providers/amazon/aws/operators/test_rds.py
@@ -18,6 +18,7 @@
 from __future__ import annotations
 
 import logging
+from typing import TYPE_CHECKING
 from unittest import mock
 from unittest.mock import patch
 
@@ -26,7 +27,6 @@
 
 from airflow.exceptions import TaskDeferred
 from airflow.models import DAG
-from airflow.providers.amazon.aws.hooks.base_aws import AwsGenericHook
 from airflow.providers.amazon.aws.hooks.rds import RdsHook
 from airflow.providers.amazon.aws.operators.rds import (
     RdsBaseOperator,
@@ -45,6 +45,9 @@
 from airflow.providers.amazon.aws.triggers.rds import RdsDbAvailableTrigger, RdsDbStoppedTrigger
 from airflow.utils import timezone
 
+if TYPE_CHECKING:
+    from airflow.providers.amazon.aws.hooks.base_aws import AwsGenericHook
+
 DEFAULT_DATE = timezone.datetime(2019, 1, 1)
 
 AWS_CONN = "amazon_default"
diff --git a/tests/providers/amazon/aws/operators/test_sagemaker_pipeline.py b/tests/providers/amazon/aws/operators/test_sagemaker_pipeline.py
index 2d509625c5db7..7e7faa5902631 100644
--- a/tests/providers/amazon/aws/operators/test_sagemaker_pipeline.py
+++ b/tests/providers/amazon/aws/operators/test_sagemaker_pipeline.py
@@ -17,8 +17,8 @@
 # under the License.
 from __future__ import annotations
 
+from typing import TYPE_CHECKING
 from unittest import mock
-from unittest.mock import MagicMock
 
 import pytest
 
@@ -30,6 +30,9 @@
 )
 from airflow.providers.amazon.aws.triggers.sagemaker import SageMakerPipelineTrigger
 
+if TYPE_CHECKING:
+    from unittest.mock import MagicMock
+
 
 class TestSageMakerStartPipelineOperator:
     @mock.patch.object(SageMakerHook, "start_pipeline")
diff --git a/tests/providers/amazon/aws/sensors/test_emr_base.py b/tests/providers/amazon/aws/sensors/test_emr_base.py
index f6b4351833989..dfeefbb63cad9 100644
--- a/tests/providers/amazon/aws/sensors/test_emr_base.py
+++ b/tests/providers/amazon/aws/sensors/test_emr_base.py
@@ -17,11 +17,15 @@
 # under the License.
 from __future__ import annotations
 
+from typing import TYPE_CHECKING
+
 import pytest
 
 from airflow.exceptions import AirflowException
 from airflow.providers.amazon.aws.sensors.emr import EmrBaseSensor
-from airflow.utils.context import Context
+
+if TYPE_CHECKING:
+    from airflow.utils.context import Context
 
 TARGET_STATE = "TARGET_STATE"
 FAILED_STATE = "FAILED_STATE"
diff --git a/tests/providers/amazon/aws/triggers/test_base.py b/tests/providers/amazon/aws/triggers/test_base.py
index aad9ce45ec4a7..9334e555b9f33 100644
--- a/tests/providers/amazon/aws/triggers/test_base.py
+++ b/tests/providers/amazon/aws/triggers/test_base.py
@@ -16,14 +16,17 @@
 # under the License.
 from __future__ import annotations
 
+from typing import TYPE_CHECKING
 from unittest import mock
 from unittest.mock import MagicMock
 
 import pytest
 
-from airflow.providers.amazon.aws.hooks.base_aws import AwsGenericHook
 from airflow.providers.amazon.aws.triggers.base import AwsBaseWaiterTrigger
-from airflow.triggers.base import TriggerEvent
+
+if TYPE_CHECKING:
+    from airflow.providers.amazon.aws.hooks.base_aws import AwsGenericHook
+    from airflow.triggers.base import TriggerEvent
 
 
 class TestImplem(AwsBaseWaiterTrigger):
diff --git a/tests/providers/amazon/aws/triggers/test_ecs.py b/tests/providers/amazon/aws/triggers/test_ecs.py
index e897bce740075..4b0a58a84a1ea 100644
--- a/tests/providers/amazon/aws/triggers/test_ecs.py
+++ b/tests/providers/amazon/aws/triggers/test_ecs.py
@@ -16,6 +16,7 @@
 # under the License.
 from __future__ import annotations
 
+from typing import TYPE_CHECKING
 from unittest import mock
 from unittest.mock import AsyncMock
 
@@ -30,7 +31,9 @@
     ClusterInactiveTrigger,
     TaskDoneTrigger,
 )
-from airflow.triggers.base import TriggerEvent
+
+if TYPE_CHECKING:
+    from airflow.triggers.base import TriggerEvent
 
 
 class TestTaskDoneTrigger:
diff --git a/tests/providers/amazon/aws/utils/eks_test_utils.py b/tests/providers/amazon/aws/utils/eks_test_utils.py
index 7a403eb855740..39201204589ad 100644
--- a/tests/providers/amazon/aws/utils/eks_test_utils.py
+++ b/tests/providers/amazon/aws/utils/eks_test_utils.py
@@ -17,12 +17,9 @@
 # under the License.
 from __future__ import annotations
 
-import datetime
 import re
 from copy import deepcopy
-from typing import Pattern, Type, Union
-
-from airflow.providers.amazon.aws.hooks.eks import EksHook
+from typing import TYPE_CHECKING, Pattern, Type, Union
 
 from ..utils.eks_test_constants import (
     STATUS,
@@ -35,6 +32,11 @@
     ResponseAttributes,
 )
 
+if TYPE_CHECKING:
+    import datetime
+
+    from airflow.providers.amazon.aws.hooks.eks import EksHook
+
 InputTypes = Union[Type[ClusterInputs], Type[NodegroupInputs], Type[FargateProfileInputs]]