Skip to content

Storage providers - GCS - Add artifact info calculation timeout #339

Description

@Ark-kun

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

Metadata

Metadata

Assignees

Labels

No labels
No labels

Type

No type

Projects

No projects

Milestone

No milestone

Relationships

None yet

Development

No branches or pull requests

Issue actions