Repository: googleapis/python-storage

Issue:
ReadTimeout errors
I keep receiving `requests.exceptions.ReadTimeout: HTTPSConnectionPool(host='storage.googleapis.com', port=443): Read timed out. (read timeout=60)` when trying to write files to storage.

I've tried adding the `timeout` value when opening the file:
`Client().bucket('my_bucket').blob('my_blob').open(mode='wb', timeout=3600)`

Investigating the traceback:
```
 File "/usr/local/lib/python3.9/site-packages/google/cloud/storage/fileio.py", line 411, in _upload_chunks_from_buffer
    upload.transmit_next_chunk(transport)
```

I've found that the method [`transmit_next_chunk`](https://github.com/googleapis/google-resumable-media-python/blob/e8219d54faa5fdec2db0fd930eb4ec719edb6f6e/google/resumable_media/requests/upload.py#L424) can receive a `timeout` param that is not being set.

Discussion:

Patch:
diff --git a/google/cloud/storage/fileio.py b/google/cloud/storage/fileio.py
index 639e731..97d2349 100644
--- a/google/cloud/storage/fileio.py
+++ b/google/cloud/storage/fileio.py
@@ -406,9 +406,15 @@ class BlobWriter(io.BufferedIOBase):
 
         upload, transport = self._upload_and_transport
 
+        # Attach timeout if specified in the keyword arguments.
+        # Otherwise, the default timeout will be used from the media library.
+        kwargs = {}
+        if "timeout" in self._upload_kwargs:
+            kwargs = {"timeout": self._upload_kwargs.get("timeout")}
+
         # Upload chunks. The SlidingBuffer class will manage seek position.
         for _ in range(num_chunks):
-            upload.transmit_next_chunk(transport)
+            upload.transmit_next_chunk(transport, **kwargs)
 
         # Wipe the buffer of chunks uploaded, preserving any remaining data.
         self._buffer.flush()


Test patch:
diff --git a/tests/unit/test_fileio.py b/tests/unit/test_fileio.py
index a92df37..cafc65e 100644
--- a/tests/unit/test_fileio.py
+++ b/tests/unit/test_fileio.py
@@ -346,6 +346,7 @@ class TestBlobWriterBinary(unittest.TestCase, _BlobWriterBase):
         blob = mock.Mock()
         upload = mock.Mock()
         transport = mock.Mock()
+        timeout = 600
 
         blob._initiate_resumable_upload.return_value = (upload, transport)
 
@@ -354,7 +355,10 @@ class TestBlobWriterBinary(unittest.TestCase, _BlobWriterBase):
             # arguments are used.
             # It would be normal to use a context manager here, but not doing so
             # gives us more control over close() for test purposes.
-            upload_kwargs = {"if_metageneration_match": 1}
+            upload_kwargs = {
+                "if_metageneration_match": 1,
+                "timeout": timeout,
+            }
             chunk_size = 8  # Note: Real upload requires a multiple of 256KiB.
             writer = self._make_blob_writer(
                 blob,
@@ -366,7 +370,7 @@ class TestBlobWriterBinary(unittest.TestCase, _BlobWriterBase):
 
         # The transmit_next_chunk method must actually consume bytes from the
         # sliding buffer for the flush() feature to work properly.
-        upload.transmit_next_chunk.side_effect = lambda _: writer._buffer.read(
+        upload.transmit_next_chunk.side_effect = lambda _, timeout: writer._buffer.read(
             chunk_size
         )
 
@@ -388,7 +392,7 @@ class TestBlobWriterBinary(unittest.TestCase, _BlobWriterBase):
             retry=None,
             **upload_kwargs
         )
-        upload.transmit_next_chunk.assert_called_with(transport)
+        upload.transmit_next_chunk.assert_called_with(transport, timeout=timeout)
         self.assertEqual(upload.transmit_next_chunk.call_count, 4)
 
         # Write another byte, finalize and close.