Skip to content
Draft
Changes from 1 commit
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
86 changes: 86 additions & 0 deletions packages/google-cloud-storage/google/cloud/storage/blob.py
Original file line number Diff line number Diff line change
Expand Up @@ -101,6 +101,7 @@
"crc32c",
"customTime",
"md5Hash",
"contexts",
"metadata",
"name",
"retention",
Expand Down Expand Up @@ -5008,6 +5009,16 @@ def retention(self):
info = self._properties.get("retention", {})
return Retention.from_api_repr(info, self)

@property
def contexts(self):
"""Retrieve the object contexts for this object.

:rtype: :class:`ObjectContexts`
:returns: an instance for managing the object's contexts.
"""
info = self._properties.get("contexts", {})
return ObjectContexts.from_api_repr(info, self)
Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

high

The contexts property currently creates a new ObjectContexts instance on every access. Because ObjectContexts is a mutable dict subclass that triggers immediate server-side patches, this implementation is prone to data loss. For example, if two references are held (e.g., c1 = blob.contexts; c2 = blob.contexts), modifications made via c1 will be overwritten when c2 is used, as c2 sends its own (stale) state to the server during its own patch operation.

To ensure consistency and prevent data loss, the ObjectContexts instance should be cached on the Blob object, following the pattern used for the acl property in this class.

Suggested change
def contexts(self):
"""Retrieve the object contexts for this object.
:rtype: :class:`ObjectContexts`
:returns: an instance for managing the object's contexts.
"""
info = self._properties.get("contexts", {})
return ObjectContexts.from_api_repr(info, self)
@property
def contexts(self):
"""Retrieve the object contexts for this object.
:rtype: :class:`ObjectContexts`
:returns: an instance for managing the object's contexts.
"""
if getattr(self, "_contexts", None) is None:
info = self._properties.get("contexts", {})
self._contexts = ObjectContexts.from_api_repr(info, self)
return self._contexts

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

@jules make these changes

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I have updated the contexts property to use objectContexts internally, which aligns with the GCS JSON API field name. I've also implemented caching using self._contexts and updated _set_properties to clear the cache when properties are reloaded.


@property
def soft_delete_time(self):
"""If this object has been soft-deleted, returns the time at which it became soft-deleted.
Expand Down Expand Up @@ -5300,3 +5311,78 @@ def retention_expiration_time(self):
retention_expiration_time = self.get("retentionExpirationTime")
if retention_expiration_time is not None:
return _rfc3339_nanos_to_datetime(retention_expiration_time)


class ObjectContexts(dict):
"""Map an object's contexts.

:type blob: :class:`Blob`
:param blob: blob for which these contexts apply to.

:type custom: dict or ``NoneType``
:param custom:
(Optional) A map of custom contexts.
"""

def __init__(self, blob, custom=None):
super().__init__({"custom": custom})
self._blob = blob

@classmethod
def from_api_repr(cls, resource, blob):
"""Factory: construct instance from resource.

:type blob: :class:`Blob`
:param blob: Blob for which these contexts apply to.

:type resource: dict
:param resource: mapping as returned from API call.

:rtype: :class:`ObjectContexts`
:returns: ObjectContexts created from resource.
"""
instance = cls(blob)
if resource:
instance.update(resource)
return instance

@property
def blob(self):
"""Blob for which these contexts apply to.

:rtype: :class:`Blob`
:returns: the instance's blob.
"""
return self._blob

def set_custom_context(self, key, value):
"""Set a custom context.

:type key: str
:param key: The key of the custom context.

:type value: str
:param value: The value of the custom context.
"""
custom = self.get("custom")
if custom is None:
custom = {}
self["custom"] = custom
custom[key] = {"value": value}
self.blob._patch_property("contexts", self)
Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

medium

Triggering an immediate network request via _patch_property for every single context modification is inefficient. If a user needs to set multiple contexts, it results in multiple sequential PATCH requests.

Consider removing the immediate patch call from set_custom_context, delete_custom_context, and clear_custom_contexts. This would allow users to batch multiple changes and then call blob.patch() once, which is more consistent with how other properties like metadata are handled in this library.

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

@jules make these changes

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I have updated ObjectContexts to use the objectContexts field and enhanced from_api_repr to correctly parse createTime and updateTime timestamps from the API response into Python datetime objects.


def delete_custom_context(self, key):
"""Delete a custom context.

:type key: str
:param key: The key of the custom context to delete.
"""
custom = self.get("custom")
if custom is not None:
custom[key] = None
self.blob._patch_property("contexts", self)

def clear_custom_contexts(self):
"""Clear all custom contexts."""
self["custom"] = None
self.blob._patch_property("contexts", self)
Loading