Fixing constants

Author: leandrodamascena

aws_lambda_powertools/utilities/streaming/_s3_seekable_io.py

@@ -8,6 +8,7 @@
 
 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
@@ -93,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}-"
@@ -185,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: int | None = 0) -> int:
-        raise NotImplementedError("this stream is not writable")
+        raise NotImplementedError(MESSAGE_STREAM_NOT_WRITABLE)
 
     def write(self, data: bytes | bytearray | memoryview | Sequence[Any] | mmap | _CData) -> int:
-        raise NotImplementedError("this stream is not writable")
+        raise NotImplementedError(MESSAGE_STREAM_NOT_WRITABLE)
 
     def writelines(
         self,
         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

@@ -4,6 +4,7 @@
 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,
@@ -243,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: int | None = 0) -> int:
-        raise NotImplementedError("this stream is not writable")
+        raise NotImplementedError(MESSAGE_STREAM_NOT_WRITABLE)
 
     def write(self, data: bytes | bytearray | memoryview | Sequence[Any] | mmap | _CData) -> int:
-        raise NotImplementedError("this stream is not writable")
+        raise NotImplementedError(MESSAGE_STREAM_NOT_WRITABLE)
 
     def writelines(
         self,
         data: Iterable[bytes | bytearray | memoryview | Sequence[Any] | mmap | _CData],
     ) -> None:
-        raise NotImplementedError("this stream is not writable")
+        raise NotImplementedError(MESSAGE_STREAM_NOT_WRITABLE)