pyright fixes

Author: nateprewitt

packages/aws-sdk-signers/src/aws_sdk_signers/_http.py

@@ -347,8 +347,8 @@ def __init__(
         self.fields = fields
 
     def __deepcopy__(
-        self, memo: dict[int, interfaces_http.Request] | None = None
-    ) -> interfaces_http.Request:
+        self, memo: dict[int, AWSRequest] | None = None
+    ) -> AWSRequest:
         if memo is None:
             memo = {}
 
@@ -358,7 +358,7 @@ def __deepcopy__(
         # the destination doesn't need to be copied because it's immutable
         # the body can't be copied because it's an iterator
         new_instance = self.__class__(
-            destination=self.destination,
+            destination=self.destination,  # pyright: ignore [reportArgumentType]
             body=self.body,
             method=self.method,
             fields=deepcopy(self.fields, memo),

packages/aws-sdk-signers/src/aws_sdk_signers/_io.py

@@ -39,7 +39,7 @@ def __init__(self, data: StreamingBlob):
         else:
             self._data = data
 
-    async def read(self, size: int = -1) -> bytes:
+    async def read(self, size: int | None = -1) -> bytes:
         """Read a number of bytes from the stream.
 
         :param size: The maximum number of bytes to read. If less than 0, all bytes will
@@ -48,14 +48,17 @@ async def read(self, size: int = -1) -> bytes:
         if self._closed or not self._data:
             raise ValueError("I/O operation on closed file.")
 
-        if isinstance(self._data, ByteStream) and not iscoroutinefunction(
+        if size is None:
+            size = -1
+
+        if isinstance(self._data, ByteStream) and not iscoroutinefunction(  # type: ignore - TODO(pyright)
             self._data.read
         ):
             # Python's runtime_checkable can't actually tell the difference between
             # sync and async, so we have to check ourselves.
             return self._data.read(size)
 
-        if isinstance(self._data, AsyncByteStream):
+        if isinstance(self._data, AsyncByteStream):  # type: ignore - TODO(pyright)
             return await self._data.read(size)
 
         return await self._read_from_iterable(
@@ -135,7 +138,7 @@ def __init__(self, data: StreamingBlob):
         if isinstance(data, bytes | bytearray):
             self._buffer = BytesIO(data)
             self._data_source = None
-        elif isinstance(data, AsyncByteStream) and iscoroutinefunction(data.read):
+        elif isinstance(data, AsyncByteStream) and iscoroutinefunction(data.read):  # type: ignore - TODO(pyright)
             # Note that we need that iscoroutine check because python won't actually check
             # whether or not the read function is async.
             self._buffer = BytesIO()
@@ -144,12 +147,15 @@ def __init__(self, data: StreamingBlob):
             self._buffer = BytesIO()
             self._data_source = AsyncBytesReader(data)
 
-    async def read(self, size: int = -1) -> bytes:
+    async def read(self, size: int | None = -1) -> bytes:
         """Read a number of bytes from the stream.
 
         :param size: The maximum number of bytes to read. If less than 0, all bytes will
             be read.
         """
+        if size is None:
+            size = -1
+
         if self._data_source is None or size == 0:
             return self._buffer.read(size)
 

packages/aws-sdk-signers/src/aws_sdk_signers/interfaces/http.py

@@ -6,7 +6,7 @@
 from collections import OrderedDict
 from collections.abc import AsyncIterable, Iterable, Iterator
 from enum import Enum
-from typing import Protocol
+from typing import Protocol, runtime_checkable
 
 
 class FieldPosition(Enum):
@@ -123,6 +123,7 @@ class Request(Protocol):
     body: AsyncIterable[bytes] | Iterable[bytes] | None
 
 
+@runtime_checkable
 class URI(Protocol):
     """Universal Resource Identifier, target location for a :py:class:`Request`."""
 

packages/aws-sdk-signers/src/aws_sdk_signers/interfaces/io.py

@@ -8,11 +8,20 @@
 class ByteStream(Protocol):
     """A file-like object with a read method that returns bytes."""
 
-    def read(self, size: int = -1) -> bytes: ...
+    def read(self, size: int | None = -1, /) -> bytes: ...
 
 
 @runtime_checkable
 class AsyncByteStream(Protocol):
     """A file-like object with an async read method."""
 
-    async def read(self, size: int = -1) -> bytes: ...
+    async def read(self, size: int | None = -1, /) -> bytes: ...
+
+
+@runtime_checkable
+class Seekable(Protocol):
+    """A file-like object with seek and tell implemented."""
+
+    def seek(self, offset: int, whence: int = 0, /) -> int: ...
+
+    def tell(self) -> int: ...

packages/aws-sdk-signers/src/aws_sdk_signers/signers.py

@@ -5,12 +5,13 @@
 import hmac
 import io
 import warnings
-from collections.abc import AsyncIterable
+from collections.abc import AsyncIterable, Iterable
 from copy import deepcopy
 from hashlib import sha256
 from typing import Required, TypedDict
 from urllib.parse import parse_qsl, quote
 
+from .interfaces.io import Seekable
 from ._http import URI, AWSRequest, Field
 from ._identity import AWSCredentialIdentity
 from ._io import AsyncBytesReader
@@ -64,6 +65,8 @@ def sign(
         new_signing_properties = self._normalize_signing_properties(
             signing_properties=signing_properties
         )
+        assert "date" in new_signing_properties
+
         new_request = self._generate_new_request(request=request)
         self._apply_required_fields(
             request=new_request,
@@ -133,14 +136,14 @@ def _signature(
         service. The date, region, service and resulting signing key are individually
         hashed, then the composite hash is used to sign the string to sign.
         """
-        assert signing_properties["date"] is not None
 
         # Components of Signing Key Calculation
         #
         # DateKey              = HMAC-SHA256("AWS4"+"<SecretAccessKey>", "<YYYYMMDD>")
         # DateRegionKey        = HMAC-SHA256(<DateKey>, "<aws-region>")
         # DateRegionServiceKey = HMAC-SHA256(<DateRegionKey>, "<aws-service>")
         # SigningKey = HMAC-SHA256(<DateRegionServiceKey>, "aws4_request")
+        assert "date" in signing_properties
         k_date = self._hash(
             key=f"AWS4{secret_key}".encode(), value=signing_properties["date"][0:8]
         )
@@ -155,7 +158,7 @@ def _hash(self, key: bytes, value: str) -> bytes:
 
     def _validate_identity(self, *, identity: AWSCredentialIdentity) -> None:
         """Perform runtime and expiration checks before attempting signing."""
-        if not isinstance(identity, AWSCredentialIdentity):
+        if not isinstance(identity, AWSCredentialIdentity): #pyright: ignore
             raise ValueError(
                 "Received unexpected value for identity parameter. Expected "
                 f"AWSCredentialIdentity but received {type(identity)}."
@@ -171,20 +174,14 @@ def _normalize_signing_properties(
     ) -> SigV4SigningProperties:
         # Create copy of signing properties to avoid mutating the original
         new_signing_properties = SigV4SigningProperties(**signing_properties)
-        new_signing_properties["date"] = self._resolve_signing_date(
-            date=new_signing_properties.get("date")
-        )
+        if "date" not in new_signing_properties:
+            date_obj = datetime.datetime.now(datetime.UTC)
+            new_signing_properties["date"] = date_obj.strftime(SIGV4_TIMESTAMP_FORMAT)
         return new_signing_properties
 
     def _generate_new_request(self, *, request: AWSRequest) -> AWSRequest:
         return deepcopy(request)
 
-    def _resolve_signing_date(self, *, date: str | None) -> str:
-        if date is None:
-            date_obj = datetime.datetime.now(datetime.UTC)
-            date = date_obj.strftime(SIGV4_TIMESTAMP_FORMAT)
-        return date
-
     def _apply_required_fields(
         self,
         *,
@@ -194,6 +191,7 @@ def _apply_required_fields(
     ) -> None:
         # Apply required X-Amz-Date if neither X-Amz-Date nor Date are present.
         if "Date" not in request.fields and "X-Amz-Date" not in request.fields:
+            assert "date" in signing_properties
             request.fields.set_field(
                 Field(name="X-Amz-Date", values=[signing_properties["date"]])
             )
@@ -283,6 +281,7 @@ def string_to_sign(
         )
 
     def _scope(self, signing_properties: SigV4SigningProperties) -> str:
+        assert "date" in signing_properties
         formatted_date = signing_properties["date"][0:8]
         region = signing_properties["region"]
         service = signing_properties["service"]
@@ -315,13 +314,13 @@ def _normalize_signing_fields(self, *, request: AWSRequest) -> dict[str, str]:
         }
         if "host" not in normalized_fields:
             normalized_fields["host"] = self._normalize_host_field(
-                uri=request.destination
+                uri=request.destination  # type: ignore - TODO(pyright)
             )
 
         return dict(sorted(normalized_fields.items()))
 
-    def _is_signable_header(self, field):
-        if field in HEADERS_EXCLUDED_FROM_SIGNING:
+    def _is_signable_header(self, field_name: str):
+        if field_name in HEADERS_EXCLUDED_FROM_SIGNING:
             return False
         return True
 
@@ -355,12 +354,6 @@ def _format_canonical_payload(
         request: AWSRequest,
         signing_properties: SigV4SigningProperties,
     ) -> str:
-        if isinstance(request.body, AsyncIterable):
-            raise TypeError(
-                "An async body was attached to a synchronous signer. Please use "
-                "AsyncSigV4Signer for async AWSRequests or ensure your body is "
-                "of type Iterable[bytes]."
-            )
         payload_hash = self._compute_payload_hash(
             request=request, signing_properties=signing_properties
         )
@@ -383,21 +376,28 @@ def _compute_payload_hash(
         if body is None:
             return EMPTY_SHA256_HASH
 
+        if not isinstance(body, Iterable):
+            raise TypeError(
+                "An async body was attached to a synchronous signer. Please use "
+                "AsyncSigV4Signer for async AWSRequests or ensure your body is "
+                "of type Iterable[bytes]."
+            )
+
         warnings.warn(
             "Payload signing is enabled. This may result in "
             "decreased performance for large request bodies.",
             AWSSDKWarning,
         )
 
         checksum = sha256()
-        if hasattr(body, "seek") and hasattr(body, "tell"):
+        if isinstance(body, Seekable):
             position = body.tell()
-            for chunk in body:  # type: ignore[union-attr]
+            for chunk in body:
                 checksum.update(chunk)
             body.seek(position)
         else:
             buffer = io.BytesIO()
-            for chunk in body:  # type: ignore[union-attr]
+            for chunk in body:
                 buffer.write(chunk)
                 checksum.update(chunk)
             buffer.seek(0)
@@ -498,14 +498,14 @@ async def _signature(
         service. The date, region, service and resulting signing key are individually
         hashed, then the composite hash is used to sign the string to sign.
         """
-        assert signing_properties.get("date") is not None
 
         # Components of Signing Key Calculation
         #
         # DateKey              = HMAC-SHA256("AWS4"+"<SecretAccessKey>", "<YYYYMMDD>")
         # DateRegionKey        = HMAC-SHA256(<DateKey>, "<aws-region>")
         # DateRegionServiceKey = HMAC-SHA256(<DateRegionKey>, "<aws-service>")
         # SigningKey = HMAC-SHA256(<DateRegionServiceKey>, "aws4_request")
+        assert "date" in signing_properties
         k_date = await self._hash(
             key=f"AWS4{secret_key}".encode(), value=signing_properties["date"][0:8]
         )
@@ -521,7 +521,7 @@ async def _hash(self, key: bytes, value: str) -> bytes:
 
     async def _validate_identity(self, *, identity: AWSCredentialIdentity) -> None:
         """Perform runtime and expiration checks before attempting signing."""
-        if not isinstance(identity, AWSCredentialIdentity):
+        if not isinstance(identity, AWSCredentialIdentity):  #pyright: ignore
             raise ValueError(
                 "Received unexpected value for identity parameter. Expected "
                 f"AWSCredentialIdentity but received {type(identity)}."
@@ -537,20 +537,14 @@ async def _normalize_signing_properties(
     ) -> SigV4SigningProperties:
         # Create copy of signing properties to avoid mutating the original
         new_signing_properties = SigV4SigningProperties(**signing_properties)
-        new_signing_properties["date"] = await self._resolve_signing_date(
-            date=new_signing_properties.get("date")
-        )
+        if "date" not in new_signing_properties:
+            date_obj = datetime.datetime.now(datetime.UTC)
+            new_signing_properties["date"] = date_obj.strftime(SIGV4_TIMESTAMP_FORMAT)
         return new_signing_properties
 
     async def _generate_new_request(self, *, request: AWSRequest) -> AWSRequest:
         return deepcopy(request)
 
-    async def _resolve_signing_date(self, *, date: str | None) -> str:
-        if date is None:
-            date_obj = datetime.datetime.now(datetime.UTC)
-            date = date_obj.strftime(SIGV4_TIMESTAMP_FORMAT)
-        return date
-
     async def _apply_required_fields(
         self,
         *,
@@ -560,6 +554,7 @@ async def _apply_required_fields(
     ) -> None:
         # Apply required X-Amz-Date if neither X-Amz-Date nor Date are present.
         if "Date" not in request.fields and "X-Amz-Date" not in request.fields:
+            assert "date" in signing_properties
             request.fields.set_field(
                 Field(name="X-Amz-Date", values=[signing_properties["date"]])
             )
@@ -654,6 +649,7 @@ async def string_to_sign(
         )
 
     async def _scope(self, signing_properties: SigV4SigningProperties) -> str:
+        assert "date" in signing_properties
         formatted_date = signing_properties["date"][0:8]
         region = signing_properties["region"]
         service = signing_properties["service"]
@@ -686,13 +682,13 @@ async def _normalize_signing_fields(self, *, request: AWSRequest) -> dict[str, s
         }
         if "host" not in normalized_fields:
             normalized_fields["host"] = await self._normalize_host_field(
-                uri=request.destination
+                uri=request.destination  # type: ignore - TODO(pyright)
             )
 
         return dict(sorted(normalized_fields.items()))
 
-    def _is_signable_header(self, field):
-        if field in HEADERS_EXCLUDED_FROM_SIGNING:
+    def _is_signable_header(self, field_name: str):
+        if field_name in HEADERS_EXCLUDED_FROM_SIGNING:
             return False
         return True
 
@@ -748,7 +744,7 @@ async def _compute_payload_hash(
         if body is None:
             return EMPTY_SHA256_HASH
 
-        if not isinstance(request.body, AsyncIterable):
+        if not isinstance(body, AsyncIterable):
             raise TypeError(
                 "A sync body was attached to an asynchronous signer. Please use "
                 "SigV4Signer for sync AWSRequests or ensure your body is "
@@ -761,14 +757,14 @@ async def _compute_payload_hash(
         )
 
         checksum = sha256()
-        if hasattr(body, "seek") and hasattr(body, "tell"):
+        if isinstance(body, Seekable):
             position = body.tell()
-            async for chunk in body:  # type: ignore[union-attr]
+            async for chunk in body:
                 checksum.update(chunk)
             body.seek(position)
         else:
             buffer = io.BytesIO()
-            async for chunk in body:  # type: ignore[union-attr]
+            async for chunk in body:
                 buffer.write(chunk)
                 checksum.update(chunk)
             buffer.seek(0)
@@ -784,7 +780,7 @@ def _remove_dot_segments(path: str, remove_consecutive_slashes: bool = True) ->
     :param remove_consecutive_slashes: Whether to remove consecutive slashes.
     :returns: The path with dot segments removed.
     """
-    output = []
+    output: list[str] = []
     for segment in path.split("/"):
         if segment == ".":
             continue