refactor(streaming): add from __future__ import annotations (#4987)

* refactor(streaming): add from __future__ import annotations

and update code according to ruff rules TCH, UP006, UP007, UP037 and
FA100.

* Fixing constants

* Organize types

---------

Co-authored-by: Leandro Damascena <lcdama@amazon.pt>

Author: ericbn

aws_lambda_powertools/utilities/streaming/_s3_seekable_io.py

@@ -1,22 +1,14 @@
+from __future__ import annotations
+
 import io
 import logging
-from typing import (
-    IO,
-    TYPE_CHECKING,
-    Any,
-    Iterable,
-    List,
-    Optional,
-    Sequence,
-    TypeVar,
-    Union,
-    cast,
-)
+from typing import IO, TYPE_CHECKING, Any, Iterable, Sequence, TypeVar, cast
 
 import boto3
 
 from aws_lambda_powertools.shared import user_agent
 from aws_lambda_powertools.utilities.streaming.compat import PowertoolsStreamingBody
+from aws_lambda_powertools.utilities.streaming.constants import MESSAGE_STREAM_NOT_WRITABLE
 
 if TYPE_CHECKING:
     from mmap import mmap
@@ -51,8 +43,8 @@ def __init__(
         self,
         bucket: str,
         key: str,
-        version_id: Optional[str] = None,
-        boto3_client: Optional["S3Client"] = None,
+        version_id: str | None = None,
+        boto3_client: S3Client | None = None,
         **sdk_options,
     ):
         self.bucket = bucket
@@ -65,10 +57,10 @@ def __init__(
         self._closed: bool = False
 
         # Caches the size of the object
-        self._size: Optional[int] = None
+        self._size: int | None = None
 
         self._s3_client = boto3_client
-        self._raw_stream: Optional[PowertoolsStreamingBody] = None
+        self._raw_stream: PowertoolsStreamingBody | None = None
 
         self._sdk_options = sdk_options
         self._sdk_options["Bucket"] = bucket
@@ -78,7 +70,7 @@ def __init__(
             self._sdk_options["VersionId"] = version_id
 
     @property
-    def s3_client(self) -> "S3Client":
+    def s3_client(self) -> S3Client:
         """
         Returns a boto3 S3 client
         """
@@ -102,7 +94,7 @@ def size(self) -> int:
     @property
     def raw_stream(self) -> PowertoolsStreamingBody:
         """
-        Returns the boto3 StreamingBody, starting the stream from the seeked position.
+        Returns the boto3 StreamingBody, starting the stream from the sought position.
         """
         if self._raw_stream is None:
             range_header = f"bytes={self._position}-"
@@ -152,19 +144,19 @@ def writable(self) -> bool:
     def tell(self) -> int:
         return self._position
 
-    def read(self, size: Optional[int] = -1) -> bytes:
+    def read(self, size: int | None = -1) -> bytes:
         size = None if size == -1 else size
         data = self.raw_stream.read(size)
         if data is not None:
             self._position += len(data)
         return data
 
-    def readline(self, size: Optional[int] = None) -> bytes:
+    def readline(self, size: int | None = None) -> bytes:
         data = self.raw_stream.readline(size)
         self._position += len(data)
         return data
 
-    def readlines(self, hint: int = -1) -> List[bytes]:
+    def readlines(self, hint: int = -1) -> list[bytes]:
         # boto3's StreamingResponse doesn't implement the "hint" parameter
         data = self.raw_stream.readlines()
         self._position += sum(len(line) for line in data)
@@ -194,19 +186,19 @@ def fileno(self) -> int:
         raise NotImplementedError("this stream is not backed by a file descriptor")
 
     def flush(self) -> None:
-        raise NotImplementedError("this stream is not writable")
+        raise NotImplementedError(MESSAGE_STREAM_NOT_WRITABLE)
 
     def isatty(self) -> bool:
         return False
 
-    def truncate(self, size: Optional[int] = 0) -> int:
-        raise NotImplementedError("this stream is not writable")
+    def truncate(self, size: int | None = 0) -> int:
+        raise NotImplementedError(MESSAGE_STREAM_NOT_WRITABLE)
 
-    def write(self, data: Union[bytes, Union[bytearray, memoryview, Sequence[Any], "mmap", "_CData"]]) -> int:
-        raise NotImplementedError("this stream is not writable")
+    def write(self, data: bytes | bytearray | memoryview | Sequence[Any] | mmap | _CData) -> int:
+        raise NotImplementedError(MESSAGE_STREAM_NOT_WRITABLE)
 
     def writelines(
         self,
-        data: Iterable[Union[bytes, Union[bytearray, memoryview, Sequence[Any], "mmap", "_CData"]]],
+        data: Iterable[bytes | bytearray | memoryview | Sequence[Any] | mmap | _CData],
     ) -> None:
-        raise NotImplementedError("this stream is not writable")
+        raise NotImplementedError(MESSAGE_STREAM_NOT_WRITABLE)

aws_lambda_powertools/utilities/streaming/constants.py

@@ -0,0 +1 @@
+MESSAGE_STREAM_NOT_WRITABLE = "this stream is not writable"

aws_lambda_powertools/utilities/streaming/s3_object.py

@@ -1,36 +1,23 @@
 from __future__ import annotations
 
 import io
-from typing import (
-    IO,
-    TYPE_CHECKING,
-    Any,
-    Iterable,
-    List,
-    Literal,
-    Optional,
-    Sequence,
-    TypeVar,
-    Union,
-    cast,
-    overload,
-)
+from typing import IO, TYPE_CHECKING, Any, Iterable, Literal, Sequence, 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
 from aws_lambda_powertools.utilities.streaming.transformations import (
     CsvTransform,
     GzipTransform,
 )
-from aws_lambda_powertools.utilities.streaming.transformations.base import (
-    BaseTransform,
-    T,
-)
+from aws_lambda_powertools.utilities.streaming.types import T
 
 if TYPE_CHECKING:
     from mmap import mmap
 
     from mypy_boto3_s3.client import S3Client
 
+    from aws_lambda_powertools.utilities.streaming.transformations.base import BaseTransform
+
     _CData = TypeVar("_CData")
 
 
@@ -74,10 +61,10 @@ def __init__(
         self,
         bucket: str,
         key: str,
-        version_id: Optional[str] = None,
-        boto3_client: Optional[S3Client] = None,
-        is_gzip: Optional[bool] = False,
-        is_csv: Optional[bool] = False,
+        version_id: str | None = None,
+        boto3_client: S3Client | None = None,
+        is_gzip: bool | None = False,
+        is_csv: bool | None = False,
         **sdk_options,
     ):
         self.bucket = bucket
@@ -94,14 +81,14 @@ def __init__(
         )
 
         # Stores the list of data transformations
-        self._data_transformations: List[BaseTransform] = []
+        self._data_transformations: list[BaseTransform] = []
         if is_gzip:
             self._data_transformations.append(GzipTransform())
         if is_csv:
             self._data_transformations.append(CsvTransform())
 
         # Stores the cached transformed stream
-        self._transformed_stream: Optional[IO[bytes]] = None
+        self._transformed_stream: IO[bytes] | None = None
 
     @property
     def size(self) -> int:
@@ -152,8 +139,8 @@ def transform(self, transformations: BaseTransform[T] | Sequence[BaseTransform[T
     def transform(
         self,
         transformations: BaseTransform[T] | Sequence[BaseTransform[T]],
-        in_place: Optional[bool] = False,
-    ) -> Optional[T]:
+        in_place: bool | None = False,
+    ) -> T | None:
         """
         Applies one or more data transformations to the stream.
 
@@ -241,10 +228,10 @@ def close(self):
     def read(self, size: int = -1) -> bytes:
         return self.transformed_stream.read(size)
 
-    def readline(self, size: Optional[int] = -1) -> bytes:
+    def readline(self, size: int | None = -1) -> bytes:
         return self.transformed_stream.readline()
 
-    def readlines(self, hint: int = -1) -> List[bytes]:
+    def readlines(self, hint: int = -1) -> list[bytes]:
         return self.transformed_stream.readlines(hint)
 
     def __next__(self):
@@ -257,19 +244,19 @@ def fileno(self) -> int:
         raise NotImplementedError("this stream is not backed by a file descriptor")
 
     def flush(self) -> None:
-        raise NotImplementedError("this stream is not writable")
+        raise NotImplementedError(MESSAGE_STREAM_NOT_WRITABLE)
 
     def isatty(self) -> bool:
         return False
 
-    def truncate(self, size: Optional[int] = 0) -> int:
-        raise NotImplementedError("this stream is not writable")
+    def truncate(self, size: int | None = 0) -> int:
+        raise NotImplementedError(MESSAGE_STREAM_NOT_WRITABLE)
 
-    def write(self, data: Union[bytes, Union[bytearray, memoryview, Sequence[Any], "mmap", "_CData"]]) -> int:
-        raise NotImplementedError("this stream is not writable")
+    def write(self, data: bytes | bytearray | memoryview | Sequence[Any] | mmap | _CData) -> int:
+        raise NotImplementedError(MESSAGE_STREAM_NOT_WRITABLE)
 
     def writelines(
         self,
-        data: Iterable[Union[bytes, Union[bytearray, memoryview, Sequence[Any], "mmap", "_CData"]]],
+        data: Iterable[bytes | bytearray | memoryview | Sequence[Any] | mmap | _CData],
     ) -> None:
-        raise NotImplementedError("this stream is not writable")
+        raise NotImplementedError(MESSAGE_STREAM_NOT_WRITABLE)

aws_lambda_powertools/utilities/streaming/transformations/base.py

@@ -1,7 +1,7 @@
 from abc import abstractmethod
-from typing import IO, Generic, TypeVar
+from typing import IO, Generic
 
-T = TypeVar("T", bound=IO[bytes])
+from aws_lambda_powertools.utilities.streaming.types import T
 
 
 class BaseTransform(Generic[T]):

aws_lambda_powertools/utilities/streaming/types.py

@@ -0,0 +1,3 @@
+from typing import IO, TypeVar
+
+T = TypeVar("T", bound=IO[bytes])