refactor(typing): enable TCH, UP and FA100 ruff rules

aws_lambda_powertools/__init__.py

@@ -1,5 +1,3 @@
-# -*- coding: utf-8 -*-
-
 """Top-level package for Lambda Python Powertools."""
 
 from pathlib import Path

aws_lambda_powertools/event_handler/api_gateway.py

@@ -1870,7 +1870,7 @@ def route(
 
         def register_resolver(func: Callable):
             methods = (method,) if isinstance(method, str) else method
-            logger.debug(f"Adding route using rule {rule} and methods: {','.join((m.upper() for m in methods))}")
+            logger.debug(f"Adding route using rule {rule} and methods: {','.join(m.upper() for m in methods)}")
 
             cors_enabled = self._cors_enabled if cors is None else cors
 

aws_lambda_powertools/event_handler/bedrock_agent.py

@@ -109,7 +109,7 @@ def get(  # type: ignore[override]
     ) -> Callable[[Callable[..., Any]], Callable[..., Any]]:
         security = None
 
-        return super(BedrockAgentResolver, self).get(
+        return super().get(
             rule,
             cors,
             compress,

aws_lambda_powertools/event_handler/openapi/params.py

@@ -328,7 +328,7 @@ def __init__(
         if default is not ...:
             raise AssertionError("Path parameters cannot have a default value")
 
-        super(Path, self).__init__(
+        super().__init__(
             default=default,
             default_factory=default_factory,
             annotation=annotation,

aws_lambda_powertools/event_handler/openapi/types.py

@@ -2,10 +2,12 @@
 
 import types
 from enum import Enum
-from typing import Any, Callable, Dict, Set, Type, TypedDict, Union
+from typing import TYPE_CHECKING, Any, Callable, Dict, Set, Type, TypedDict, Union
 
 from pydantic import BaseModel
-from typing_extensions import NotRequired
+
+if TYPE_CHECKING:
+    from typing_extensions import NotRequired
 
 CacheKey = Union[Callable[..., Any], None]
 IncEx = Union[Set[int], Set[str], Dict[int, Any], Dict[str, Any]]

aws_lambda_powertools/logging/types.py

@@ -1,8 +1,9 @@
 from __future__ import annotations
 
-from typing import Any, Dict, TypedDict, Union
+from typing import TYPE_CHECKING, Any, Dict, TypedDict, Union
 
-from typing_extensions import NotRequired, TypeAlias
+if TYPE_CHECKING:
+    from typing_extensions import NotRequired, TypeAlias
 
 
 class PowertoolsLogRecord(TypedDict):

aws_lambda_powertools/metrics/provider/cloudwatch_emf/types.py

@@ -1,8 +1,9 @@
 from __future__ import annotations
 
-from typing import TypedDict
+from typing import TYPE_CHECKING, TypedDict
 
-from typing_extensions import NotRequired
+if TYPE_CHECKING:
+    from typing_extensions import NotRequired
 
 
 class CloudWatchEMFMetric(TypedDict):

aws_lambda_powertools/shared/cache_dict.py

@@ -25,7 +25,7 @@ def __setitem__(self, key, value):
             del self[oldest]
 
     def get(self, key, *args, **kwargs):
-        item = super(LRUDict, self).get(key, *args, **kwargs)
+        item = super().get(key, *args, **kwargs)
         if item:
             self.move_to_end(key=key)
         return item

aws_lambda_powertools/shared/lazy_import.py

@@ -32,7 +32,7 @@ def __init__(self, local_name, parent_module_globals, name):  # pylint: disable=
         self._local_name = local_name
         self._parent_module_globals = parent_module_globals
 
-        super(LazyLoader, self).__init__(name)
+        super().__init__(name)
 
     def _load(self):
         # Import the target module and insert it into the parent's namespace

aws_lambda_powertools/utilities/__init__.py

@@ -1,3 +1 @@
-# -*- coding: utf-8 -*-
-
 """General utilities for Powertools"""

aws_lambda_powertools/utilities/batch/__init__.py

@@ -1,5 +1,3 @@
-# -*- coding: utf-8 -*-
-
 """
 Batch processing utility
 """

aws_lambda_powertools/utilities/batch/base.py

@@ -1,7 +1,7 @@
-# -*- coding: utf-8 -*-
 """
 Batch processing utilities
 """
+
 from __future__ import annotations
 
 import asyncio

aws_lambda_powertools/utilities/batch/exceptions.py

@@ -34,7 +34,7 @@ def __init__(self, msg="", child_exceptions: list[ExceptionInfo] | None = None):
         super().__init__(msg, child_exceptions)
 
     def __str__(self):
-        parent_exception_str = super(BatchProcessingError, self).__str__()
+        parent_exception_str = super().__str__()
         return self.format_exceptions(parent_exception_str)
 
 

aws_lambda_powertools/utilities/data_classes/kinesis_firehose_event.py

@@ -5,12 +5,13 @@
 import warnings
 from dataclasses import dataclass, field
 from functools import cached_property
-from typing import Any, Callable, ClassVar, Iterator
-
-from typing_extensions import Literal
+from typing import TYPE_CHECKING, Any, Callable, ClassVar, Iterator
 
 from aws_lambda_powertools.utilities.data_classes.common import DictWrapper
 
+if TYPE_CHECKING:
+    from typing_extensions import Literal
+
 
 @dataclass(repr=False, order=False, frozen=True)
 class KinesisFirehoseDataTransformationRecordMetadata:

aws_lambda_powertools/utilities/feature_flags/feature_flags.py

@@ -14,12 +14,11 @@
     compare_time_range,
 )
 from aws_lambda_powertools.utilities.feature_flags.exceptions import ConfigurationStoreError
-from aws_lambda_powertools.utilities.feature_flags.types import P, T
 
 if TYPE_CHECKING:
     from aws_lambda_powertools.logging import Logger
     from aws_lambda_powertools.utilities.feature_flags.base import StoreProvider
-    from aws_lambda_powertools.utilities.feature_flags.types import JSONType
+    from aws_lambda_powertools.utilities.feature_flags.types import JSONType, P, T
 
 
 RULE_ACTION_MAPPING = {

aws_lambda_powertools/utilities/idempotency/persistence/dynamodb.py

@@ -123,7 +123,7 @@ def __init__(
 
         self._deserializer = TypeDeserializer()
 
-        super(DynamoDBPersistenceLayer, self).__init__()
+        super().__init__()
 
     def _get_key(self, idempotency_key: str) -> dict:
         """Build primary key attribute simple or composite based on params.

aws_lambda_powertools/utilities/idempotency/persistence/redis.py

@@ -310,7 +310,7 @@ def lambda_handler(event: dict, context: LambdaContext):
         self.validation_key_attr = validation_key_attr
         self._json_serializer = json.dumps
         self._json_deserializer = json.loads
-        super(RedisCachePersistenceLayer, self).__init__()
+        super().__init__()
         self._orphan_lock_timeout = min(10, self.expires_after_seconds)
 
     def _get_expiry_second(self, expiry_timestamp: int | None = None) -> int:

aws_lambda_powertools/utilities/parameters/__init__.py

@@ -1,5 +1,3 @@
-# -*- coding: utf-8 -*-
-
 """
 Parameter retrieval and caching utility
 """

aws_lambda_powertools/utilities/typing/__init__.py

@@ -1,5 +1,3 @@
-# -*- coding: utf-8 -*-
-
 """
 Typing for developer ease in the IDE
 """

aws_lambda_powertools/utilities/typing/lambda_client_context.py

@@ -1,4 +1,3 @@
-# -*- coding: utf-8 -*-
 from __future__ import annotations
 
 from typing import TYPE_CHECKING, Any
@@ -9,7 +8,7 @@
     )
 
 
-class LambdaClientContext(object):
+class LambdaClientContext:
     _client: LambdaClientContextMobileClient
     _custom: dict[str, Any]
     _env: dict[str, Any]

aws_lambda_powertools/utilities/typing/lambda_client_context_mobile_client.py

@@ -1,7 +1,4 @@
-# -*- coding: utf-8 -*-
-
-
-class LambdaClientContextMobileClient(object):
+class LambdaClientContextMobileClient:
     """Mobile Client context that's provided to Lambda by the client application."""
 
     _installation_id: str

aws_lambda_powertools/utilities/typing/lambda_cognito_identity.py

@@ -1,7 +1,4 @@
-# -*- coding: utf-8 -*-
-
-
-class LambdaCognitoIdentity(object):
+class LambdaCognitoIdentity:
     """Information about the Amazon Cognito identity that authorized the request."""
 
     _cognito_identity_id: str

aws_lambda_powertools/utilities/typing/lambda_context.py

@@ -1,4 +1,3 @@
-# -*- coding: utf-8 -*-
 from __future__ import annotations
 
 from typing import TYPE_CHECKING
@@ -12,7 +11,7 @@
     )
 
 
-class LambdaContext(object):
+class LambdaContext:
     """The LambdaContext static object can be used to ease the development by providing the IDE type hints.
 
     Example

docs/utilities/batch.md

@@ -502,7 +502,7 @@ Use the context manager to access a list of all returned values from your `recor
 
 === "Accessing raw processed messages"
 
-    ```python hl_lines="29-36"
+    ```python hl_lines="28-35"
     --8<-- "examples/batch_processing/src/context_manager_access.py"
     ```
 

examples/batch_processing/src/context_manager_access.py

@@ -1,7 +1,6 @@
 from __future__ import annotations
 
 import json
-from typing import List, Tuple
 
 from typing_extensions import Literal
 
@@ -28,7 +27,7 @@ def record_handler(record: SQSRecord):
 def lambda_handler(event, context: LambdaContext):
     batch = event["Records"]  # (1)!
     with processor(records=batch, handler=record_handler):
-        processed_messages: List[Tuple] = processor.process()
+        processed_messages: list[tuple] = processor.process()
 
     for message in processed_messages:
         status: Literal["success", "fail"] = message[0]

examples/idempotency/src/bring_your_own_persistent_store.py

@@ -36,7 +36,7 @@ def __init__(
         self.status_attr = status_attr
         self.data_attr = data_attr
         self.validation_key_attr = validation_key_attr
-        super(MyOwnPersistenceLayer, self).__init__()
+        super().__init__()
 
     def _item_to_data_record(self, item: Dict[str, Any]) -> DataRecord:
         """

ruff.toml

@@ -23,7 +23,9 @@ lint.select = [
     "Q",   # flake8-quotes - https://beta.ruff.rs/docs/rules/#flake8-quotes-q
     "PTH", # flake8-use-pathlib - https://beta.ruff.rs/docs/rules/#flake8-use-pathlib-pth
     "T10", # flake8-debugger https://beta.ruff.rs/docs/rules/#flake8-debugger-t10
+    "TCH", # flake8-type-checking - https://docs.astral.sh/ruff/rules/#flake8-type-checking-tch
     "TD",  # flake8-todo - https://beta.ruff.rs/docs/rules/#flake8-todos-td
+    "UP",  # pyupgrade - https://docs.astral.sh/ruff/rules/#pyupgrade-up
     "W",   # pycodestyle warning - https://beta.ruff.rs/docs/rules/#warning-w
 ]
 
@@ -36,7 +38,6 @@ lint.ignore = [
     "B904",    # raise-without-from-inside-except - disabled temporarily
     "PLC1901", # Compare-to-empty-string - disabled temporarily
     "PYI024",
-    "FA100",   # Enable this rule when drop support to Python 3.7
 ]
 
 # Exclude files and directories
@@ -58,6 +59,7 @@ exclude = [
 # Maximum line length
 line-length = 120
 
+target-version = "py38"
 
 fix = true
 lint.fixable = ["I", "COM812", "W"]
@@ -81,6 +83,9 @@ max-statements = 70
 [lint.isort]
 split-on-trailing-comma = true
 
+[lint.flake8-type-checking]
+runtime-evaluated-base-classes = ["pydantic.BaseModel"]
+
 [lint.per-file-ignores]
 # Ignore specific rules for specific files
 "tests/e2e/utils/data_builder/__init__.py" = ["F401"]
@@ -90,6 +95,6 @@ split-on-trailing-comma = true
 "aws_lambda_powertools/event_handler/openapi/compat.py" = ["F401"]
 # Maintenance: we're keeping EphemeralMetrics code in case of Hyrum's law so we can quickly revert it
 "aws_lambda_powertools/metrics/metrics.py" = ["ERA001"]
-"examples/*" = ["FA100"]
-"tests/*" = ["FA100"]
+"examples/*" = ["FA100", "TCH"]
+"tests/*" = ["FA100", "TCH"]
 "aws_lambda_powertools/utilities/parser/models/*" = ["FA100"]

tests/e2e/parser/handlers/handler_with_union_tag.py

@@ -1,5 +1,3 @@
-from __future__ import annotations
-
 from typing import Literal, Union
 
 from pydantic import BaseModel, Field

tests/functional/data_masking/test_aws_encryption_sdk.py

@@ -1,9 +1,7 @@
-from __future__ import annotations
-
 import base64
 import functools
 import json
-from typing import Any, Callable
+from typing import Any, Callable, Union
 
 import pytest
 from aws_encryption_sdk.identifiers import Algorithm
@@ -24,7 +22,7 @@ def __init__(
     ) -> None:
         super().__init__(json_serializer, json_deserializer)
 
-    def encrypt(self, data: bytes | str, **kwargs) -> str:
+    def encrypt(self, data: Union[bytes, str], **kwargs) -> str:
         encoded_data: str = self.json_serializer(data)
         ciphertext = base64.b64encode(encoded_data.encode("utf-8")).decode()
         return ciphertext

tests/functional/event_handler/test_openapi_serialization.py

@@ -48,7 +48,7 @@ def serializer(_):
     app = APIGatewayRestResolver(enable_validation=True, serializer=serializer)
 
     # GIVEN a custom class
-    class CustomClass(object):
+    class CustomClass:
         __slots__ = []
 
     # GIVEN a handler that returns an instance of that class

tests/functional/idempotency/persistence/test_redis_layer.py

@@ -100,7 +100,7 @@ def __init__(self, cache: dict = None, mock_latency_ms: int = 0, **kwargs):
         self.closed = False
         self.mock_latency_ms = mock_latency_ms
         self.nx_lock = Lock()
-        super(MockRedis, self).__init__()
+        super().__init__()
 
     # check_closed is called before every mock redis operation
     def check_closed(self):

tests/functional/idempotency/test_idempotency.py

@@ -1179,7 +1179,7 @@ def handler(event, context):
 class MockPersistenceLayer(BasePersistenceLayer):
     def __init__(self, expected_idempotency_key: str):
         self.expected_idempotency_key = expected_idempotency_key
-        super(MockPersistenceLayer, self).__init__()
+        super().__init__()
 
     def _put_record(self, data_record: DataRecord) -> None:
         assert data_record.idempotency_key == self.expected_idempotency_key

tests/functional/idempotency/utils.py

@@ -1,8 +1,6 @@
-from __future__ import annotations
-
 import hashlib
 import json
-from typing import Any, Dict
+from typing import Any, Dict, Optional
 
 from botocore import stub
 from pytest import FixtureRequest
@@ -90,7 +88,7 @@ def build_idempotency_put_item_response_stub(
     expiration: int,
     status: str,
     request: FixtureRequest,
-    validation_data: Any | None,
+    validation_data: Optional[Any],
 ):
     response = {
         "Item": {