See #338
Example implementation:
import time
import typing
import google.cloud.storage
class GcsClientWithProperDeadlines(google.cloud.storage.Client):
def __init__(self, *args, list_resources_deadline_seconds: float = 5.0, **kwargs):
super().__init__(*args, **kwargs)
self._list_resources_deadline_seconds = list_resources_deadline_seconds
# @functools.wraps(google.cloud.storage.Client.list_blobs)
def list_blobs(self, *args, **kwargs) -> typing.Iterator[google.cloud.storage.Blob]:
start_time_seconds = time.monotonic()
blob_stream = super().list_blobs(*args, **kwargs)
for blob in blob_stream:
if (
time.monotonic() - start_time_seconds
> self._list_resources_deadline_seconds
):
raise RuntimeError(
f"Timed out listing GCS blobs after {self._list_resources_deadline_seconds} seconds"
)
yield blob
See #338
Example implementation: