Add option to set If-Match eTag for objects write requests

.pubnub.yml

@@ -1,5 +1,5 @@
 name: python
-version: 10.1.0
+version: 10.2.0
 schema: 1
 scm: github.com/pubnub/python
 sdks:
@@ -18,7 +18,7 @@ sdks:
         distributions:
           - distribution-type: library
             distribution-repository: package
-            package-name: pubnub-10.1.0
+            package-name: pubnub-10.2.0
             location: https://pypi.org/project/pubnub/
             supported-platforms:
               supported-operating-systems:
@@ -91,8 +91,8 @@ sdks:
           -
             distribution-type: library
             distribution-repository: git release
-            package-name: pubnub-10.1.0
-            location: https://github.com/pubnub/python/releases/download/10.1.0/pubnub-10.1.0.tar.gz
+            package-name: pubnub-10.2.0
+            location: https://github.com/pubnub/python/releases/download/10.2.0/pubnub-10.2.0.tar.gz
             supported-platforms:
               supported-operating-systems:
                 Linux:
@@ -163,6 +163,11 @@ sdks:
                 license-url: https://github.com/encode/httpx/blob/master/LICENSE.md
                 is-required: Required
 changelog:
+  - date: 2025-02-07
+    version: 10.2.0
+    changes:
+      - type: feature
+        text: "Write protection with `If-Match` eTag header for setting channel and uuid metadata."
   - date: 2025-01-30
     version: 10.1.0
     changes:

CHANGELOG.md

@@ -1,3 +1,9 @@
+## 10.2.0
+February 07 2025
+
+#### Added
+- Write protection with `If-Match` eTag header for setting channel and uuid metadata.
+
 ## 10.1.0
 January 30 2025
 

examples/native_sync/using_etag.py

@@ -0,0 +1,60 @@
+import os
+
+from copy import deepcopy
+from pubnub.pubnub import PubNub
+from pubnub.pnconfiguration import PNConfiguration
+from pubnub.exceptions import PubNubException
+
+config = PNConfiguration()
+config.publish_key = os.getenv('PUBLISH_KEY', default='demo')
+config.subscribe_key = os.getenv('SUBSCRIBE_KEY', default='demo')
+config.user_id = "example"
+
+config_2 = deepcopy(config)
+config_2.user_id = "example_2"
+
+pubnub = PubNub(config)
+pubnub_2 = PubNub(config_2)
+
+sample_user = {
+    "uuid": "SampleUser",
+    "name": "John Doe",
+    "email": "jd@example.com",
+    "custom": {"age": 42, "address": "123 Main St."},
+}
+
+# One client creates a metada for the user "SampleUser" and successfully writes it to the server.
+set_result = pubnub.set_uuid_metadata(
+    **sample_user,
+    include_custom=True,
+    include_status=True,
+    include_type=True
+).sync()
+
+# We store the eTag for the user for further updates.
+original_e_tag = set_result.result.data.get('eTag')
+
+# Another client sets the user meta with the same UUID but different data.
+overwrite_result = pubnub_2.set_uuid_metadata(uuid="SampleUser", name="Jane Doe").sync()
+new_e_tag = overwrite_result.result.data.get('eTag')
+
+# We can verify that there is a new eTag for the user.
+print(f"{original_e_tag == new_e_tag=}")
+
+# We modify the user and try to update it.
+updated_user = {**sample_user, "custom": {"age": 43, "address": "321 Other St."}}
+
+try:
+    update_result = pubnub.set_uuid_metadata(
+        **updated_user,
+        include_custom=True,
+        include_status=True,
+        include_type=True
+    ).if_matches_etag(original_e_tag).sync()
+except PubNubException as e:
+    # We get an exception and after reading the error message we can see that the reason is that the eTag is outdated.
+    print(f"Update failed: {e.get_error_message().get('message')}\nHTTP Status Code: {e.get_status_code()}")
+
+
+except Exception as e:
+    print(f"Unexpected error: {e}")

pubnub/endpoints/endpoint.py

@@ -25,6 +25,7 @@ class Endpoint(object):
 
     __metaclass__ = ABCMeta
     _path = None
+    _custom_headers: dict = None
 
     def __init__(self, pubnub):
         self.pubnub = pubnub
@@ -100,6 +101,9 @@ def request_headers(self):
         if self.http_method() in [HttpMethod.POST, HttpMethod.PATCH]:
             headers["Content-type"] = "application/json"
 
+        if self._custom_headers:
+            headers.update(self._custom_headers)
+
         return headers
 
     def build_file_upload_request(self):

pubnub/endpoints/objects_v2/objects_endpoint.py

@@ -17,6 +17,10 @@ class ObjectsEndpoint(Endpoint):
 
     _includes: PNIncludes = None
 
+    __if_matches_etag: str = None
+
+    _custom_headers: dict = {}
+
     def __init__(self, pubnub):
         Endpoint.__init__(self, pubnub)
 
@@ -36,6 +40,11 @@ def validate_params(self):
     def validate_specific_params(self):
         pass
 
+    def if_matches_etag(self, etag: str):
+        self.__if_matches_etag = etag
+        self._custom_headers.update({"If-Match": etag})
+        return self
+
     def encoded_params(self):
         params = {}
         if isinstance(self, ListEndpoint):

pubnub/exceptions.py

@@ -1,3 +1,6 @@
+from json import loads, JSONDecodeError
+
+
 class PubNubException(Exception):
     def __init__(self, errormsg="", status_code=0, pn_error=None, status=None):
         self._errormsg = errormsg
@@ -19,6 +22,16 @@ def _status(self):
         raise DeprecationWarning
         return self.status
 
+    def get_status_code(self):
+        return self._status_code
+
+    def get_error_message(self):
+        try:
+            error = loads(self._errormsg)
+            return error.get('error')
+        except JSONDecodeError:
+            return self._errormsg
+
 
 class PubNubAsyncioException(Exception):
     def __init__(self, result, status):

pubnub/pubnub_core.py

@@ -96,7 +96,7 @@
 
 class PubNubCore:
     """A base class for PubNub Python API implementations"""
-    SDK_VERSION = "10.1.0"
+    SDK_VERSION = "10.2.0"
     SDK_NAME = "PubNub-Python"
 
     TIMESTAMP_DIVIDER = 1000
@@ -288,9 +288,9 @@ def set_uuid_metadata(self, uuid: str = None, include_custom: bool = None, custo
                        include_type=include_type, status=status, type=type, name=name, email=email,
                        external_id=external_id, profile_url=profile_url)
 
-    def get_uuid_metadata(self, uuud: str = None, include_custom: bool = None, include_status: bool = True,
+    def get_uuid_metadata(self, uuid: str = None, include_custom: bool = None, include_status: bool = True,
                           include_type: bool = True) -> GetUuid:
-        return GetUuid(self, uuid=uuud, include_custom=include_custom, include_status=include_status,
+        return GetUuid(self, uuid=uuid, include_custom=include_custom, include_status=include_status,
                        include_type=include_type)
 
     def remove_uuid_metadata(self, uuid: str = None) -> RemoveUuid:

setup.py

@@ -2,7 +2,7 @@
 
 setup(
     name='pubnub',
-    version='10.1.0',
+    version='10.2.0',
     description='PubNub Real-time push service in the cloud',
     author='PubNub',
     author_email='support@pubnub.com',