fix: addressing code review comments:

* Fixed _upload.py to check and safely .decode("utf-8") header values if they are of type bytes
* Fixed requests_upload.py to gracefully handle naive datetime deadlines by replacing timezone info with UTC.
* Re-implemented the chunk_size property in _upload.py using the "standard formula for alignment": ((size + granularity - 1) // granularity) * granularity.
* Updated the chunk transmission loop to check for state machine invalid state.
* Added `requests.exceptions.ChunkedEncodingError` to the list of default retriable exceptions

Author: Test User

packages/google-api-core/google/api_core/resumable_media/_upload.py

@@ -73,13 +73,12 @@ def invalid(self) -> bool:
     @property
     def chunk_size(self) -> int:
         """int: The block-aligned chunk size informed by server granularity."""
-        actual_chunk_size = self._chunk_size
         if self._chunk_granularity:
-            if actual_chunk_size % self._chunk_granularity != 0:
-                actual_chunk_size = (
-                    (actual_chunk_size // self._chunk_granularity) + 1
-                ) * self._chunk_granularity
-        return actual_chunk_size
+            return (
+                (self._chunk_size + self._chunk_granularity - 1)
+                // self._chunk_granularity
+            ) * self._chunk_granularity
+        return self._chunk_size
 
     def build_initiate_request(
         self,
@@ -110,7 +109,7 @@ def build_initiate_request(
         # Merge user metadata first
         if stream_metadata:
             for k, v in stream_metadata:
-                headers[k] = str(v)
+                headers[k] = v.decode("utf-8") if isinstance(v, bytes) else str(v)
 
         # Critical protocol headers overwrite user metadata
         headers[_common.UPLOAD_PROTOCOL_HEADER] = _common.UPLOAD_PROTOCOL_VALUE

packages/google-api-core/google/api_core/resumable_media/requests_upload.py

@@ -112,10 +112,18 @@ def finished(self) -> bool:
         """bool: Indicates if the upload has completed successfully."""
         return self._machine.finished
 
+    @property
+    def invalid(self) -> bool:
+        """bool: Indicates if the upload has entered an unrecoverable state."""
+        return self._machine.invalid
+
     def _check_deadline(self) -> Optional[float]:
         if self.deadline:
             now = datetime.datetime.now(datetime.timezone.utc)
-            timeout = (self.deadline - now).total_seconds()
+            deadline = self.deadline
+            if deadline.tzinfo is None:
+                deadline = deadline.replace(tzinfo=datetime.timezone.utc)
+            timeout = (deadline - now).total_seconds()
             if timeout <= 0:
                 raise exceptions.DeadlineExceeded(
                     f"Upload deadline {self.deadline} was exceeded."
@@ -128,7 +136,7 @@ def should_retry(exc):
             if isinstance(exc, requests.exceptions.RequestException):
                 if isinstance(
                     exc,
-                    (requests.exceptions.ConnectionError, requests.exceptions.Timeout),
+                    (requests.exceptions.ConnectionError, requests.exceptions.ChunkedEncodingError, requests.exceptions.Timeout),
                 ):
                     return True
             if isinstance(exc, exceptions.GoogleAPICallError):
@@ -469,7 +477,7 @@ def make_resumable_upload(
     )
 
     final_response = None
-    while not upload.finished:
+    while not upload.finished and not upload.invalid:
         final_response = upload.transmit_next_chunk(transport=transport)
     assert final_response is not None
     return final_response