refactor(streaming): use standard collections for types (#6483)

* Using generics types

* Adding shared

Author: leandrodamascena

aws_lambda_powertools/shared/dynamodb_deserializer.py

@@ -1,7 +1,10 @@
 from __future__ import annotations
 
 from decimal import Clamped, Context, Decimal, Inexact, Overflow, Rounded, Underflow
-from typing import Any, Callable, Sequence
+from typing import TYPE_CHECKING, Any
+
+if TYPE_CHECKING:
+    from collections.abc import Callable, Sequence
 
 # NOTE: DynamoDB supports up to 38 digits precision
 # Therefore, this ensures our Decimal follows what's stored in the table

aws_lambda_powertools/shared/functions.py

@@ -8,10 +8,13 @@
 import warnings
 from binascii import Error as BinAsciiError
 from pathlib import Path
-from typing import Any, Generator, overload
+from typing import TYPE_CHECKING, Any, overload
 
 from aws_lambda_powertools.shared import constants
 
+if TYPE_CHECKING:
+    from collections.abc import Generator
+
 logger = logging.getLogger(__name__)
 
 

aws_lambda_powertools/shared/types.py

@@ -1,3 +1,4 @@
-from typing import Any, Callable, TypeVar
+from collections.abc import Callable
+from typing import Any, TypeVar
 
-AnyCallableT = TypeVar("AnyCallableT", bound=Callable[..., Any])  # noqa: VNE001
+AnyCallableT = TypeVar("AnyCallableT", bound=Callable[..., Any])

aws_lambda_powertools/utilities/serialization.py

@@ -2,7 +2,8 @@
 
 import base64
 import json
-from typing import Any, Callable
+from collections.abc import Callable
+from typing import Any
 
 
 def base64_encode(data: str) -> str:

aws_lambda_powertools/utilities/streaming/_s3_seekable_io.py

@@ -2,7 +2,7 @@
 
 import io
 import logging
-from typing import IO, TYPE_CHECKING, Any, Iterable, Sequence, TypeVar, cast
+from typing import IO, TYPE_CHECKING, Any, TypeVar, cast
 
 import boto3
 
@@ -11,6 +11,7 @@
 from aws_lambda_powertools.utilities.streaming.constants import MESSAGE_STREAM_NOT_WRITABLE
 
 if TYPE_CHECKING:
+    from collections.abc import Iterable, Sequence
     from mmap import mmap
 
     from mypy_boto3_s3.client import S3Client

aws_lambda_powertools/utilities/streaming/s3_object.py

@@ -1,7 +1,8 @@
 from __future__ import annotations
 
 import io
-from typing import IO, TYPE_CHECKING, Any, Iterable, Literal, Sequence, TypeVar, cast, overload
+from collections.abc import Sequence
+from typing import IO, TYPE_CHECKING, Any, Literal, TypeVar, cast, overload
 
 from aws_lambda_powertools.utilities.streaming._s3_seekable_io import _S3SeekableIO
 from aws_lambda_powertools.utilities.streaming.constants import MESSAGE_STREAM_NOT_WRITABLE
@@ -12,6 +13,7 @@
 from aws_lambda_powertools.utilities.streaming.types import T
 
 if TYPE_CHECKING:
+    from collections.abc import Iterable
     from mmap import mmap
 
     from mypy_boto3_s3.client import S3Client

tests/functional/streaming/_boto3/test_s3_object.py

@@ -1,3 +1,5 @@
+from __future__ import annotations
+
 from csv import DictReader
 from gzip import GzipFile
 

tests/functional/streaming/_boto3/test_s3_seekable_io.py

@@ -1,3 +1,5 @@
+from __future__ import annotations
+
 import io
 
 import boto3

tests/unit/shared/test_dynamodb_deserializer.py

@@ -1,4 +1,6 @@
-from typing import Any, Dict, Optional
+from __future__ import annotations
+
+from typing import Any
 
 import pytest
 
@@ -10,14 +12,14 @@ def __init__(self, data: dict):
         self._data = data
         self._deserializer = TypeDeserializer()
 
-    def _deserialize_dynamodb_dict(self) -> Optional[Dict[str, Any]]:
+    def _deserialize_dynamodb_dict(self) -> dict[str, Any] | None:
         if self._data is None:
             return None
 
         return {k: self._deserializer.deserialize(v) for k, v in self._data.items()}
 
     @property
-    def data(self) -> Optional[Dict[str, Any]]:
+    def data(self) -> dict[str, Any] | None:
         """The primary key attribute(s) for the DynamoDB item that was modified."""
         return self._deserialize_dynamodb_dict()
 

tests/unit/test_cookie_class.py

@@ -1,3 +1,5 @@
+from __future__ import annotations
+
 from datetime import datetime
 
 from aws_lambda_powertools.shared.cookies import Cookie, SameSite