From 59b63cc7800e422bfd091875f742a042c96e6a7a Mon Sep 17 00:00:00 2001 From: sunil-lakshman <104969541+sunil-lakshman@users.noreply.github.com> Date: Thu, 14 May 2026 12:16:38 +0530 Subject: [PATCH] Added branches support in entry variants --- CHANGELOG.md | 8 ++ contentstack_management/__init__.py | 2 +- contentstack_management/entries/entry.py | 39 +++-- .../entry_variants/entry_variants.py | 136 +++++++++++++++--- .../entry_variants/test_entry_variants.py | 72 +++++++++- 5 files changed, 220 insertions(+), 37 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 568734c..62b0479 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,6 +1,14 @@ # CHANGELOG ## Content Management SDK For Python +--- +## v1.9.0 + +#### Date: 18 May 2026 + +- Entry variants: added `publish` and `unpublish` for the entry publish/unpublish endpoints; documented payloads including `entry.variants` and optional `entry.variant_rules` on publish. +- Entry variants: optional stack branch via the `branch` request header; `Entry.variants()` accepts no arguments, a branch UID only, or `(variant_uid, branch)` (use `variants(variant_uid, None)` when targeting a variant without a branch). + --- ## v1.8.1 diff --git a/contentstack_management/__init__.py b/contentstack_management/__init__.py index f416031..efadd82 100644 --- a/contentstack_management/__init__.py +++ b/contentstack_management/__init__.py @@ -82,7 +82,7 @@ __author__ = 'dev-ex' __status__ = 'debug' __region__ = 'na' -__version__ = '1.8.1' +__version__ = '1.9.0' __host__ = 'api.contentstack.io' __protocol__ = 'https://' __api_version__ = 'v3' diff --git a/contentstack_management/entries/entry.py b/contentstack_management/entries/entry.py index 3422eca..0c643e6 100644 --- a/contentstack_management/entries/entry.py +++ b/contentstack_management/entries/entry.py @@ -423,13 +423,15 @@ def unpublish(self, data): data = json.dumps(data) return self.client.post(url, headers = self.client.headers, data = data, params = self.params) - def variants(self, variant_uid: str = None): + def variants(self, *args): """ Returns an EntryVariants instance for managing variant entries. - - :param variant_uid: The `variant_uid` parameter is a string that represents the unique identifier of - the variant. It is used to specify which variant to work with - :type variant_uid: str + + Pass no arguments when no ``branch`` header is required, one positional argument for the + branch UID only (sent as the ``branch`` request header), or two + arguments ``(variant_uid, branch)`` for a specific variant on a branch. + + :param args: ``()`` | ``(branch,)`` | ``(variant_uid, branch)`` :return: EntryVariants instance for managing variant entries ------------------------------- [Example:] @@ -438,13 +440,26 @@ def variants(self, variant_uid: str = None): >>> client = contentstack_management.Client(authtoken='your_authtoken') >>> # Get all variant entries >>> result = client.stack('api_key').content_types('content_type_uid').entry('entry_uid').variants().query().find().json() - >>> # Get specific variant entry - >>> result = client.stack('api_key').content_types('content_type_uid').entry('entry_uid').variants('variant_uid').fetch().json() - - ------------------------------- - """ - - return EntryVariants(self.client, self.content_type_uid, self.entry_uid, variant_uid) + >>> # Variant operations on a branch (branch header only) + >>> result = client.stack('api_key').content_types('content_type_uid').entry('entry_uid').variants('branch_uid').query().find().json() + >>> # Specific variant on a branch + >>> result = client.stack('api_key').content_types('content_type_uid').entry('entry_uid').variants('variant_uid', 'branch_uid').fetch().json() + + ------------------------------- + """ + variant_uid = None + branch = None + if len(args) == 1: + branch = args[0] + elif len(args) == 2: + variant_uid, branch = args[0], args[1] + elif len(args) > 2: + raise TypeError( + f"variants() takes at most 2 positional arguments ({len(args)} given)" + ) + return EntryVariants( + self.client, self.content_type_uid, self.entry_uid, variant_uid, branch + ) def includeVariants(self, include_variants: str = 'true', variant_uid: str = None, params: dict = None): """ diff --git a/contentstack_management/entry_variants/entry_variants.py b/contentstack_management/entry_variants/entry_variants.py index 5dc6bb7..674ef10 100644 --- a/contentstack_management/entry_variants/entry_variants.py +++ b/contentstack_management/entry_variants/entry_variants.py @@ -1,28 +1,49 @@ """This class takes a base URL as an argument when it's initialized, which is the endpoint for the RESTFUL API that we'll be interacting with. -The query(), create(), fetch(), delete(), update(), versions(), and includeVariants() methods each correspond to +The query(), create(), fetch(), delete(), update(), versions(), includeVariants(), publish(), and unpublish() methods each correspond to the operations that can be performed on the API """ import json from ..common import Parameter from .._errors import ArgumentException -from .._messages import ENTRY_VARIANT_CONTENT_TYPE_UID_REQUIRED, ENTRY_VARIANT_ENTRY_UID_REQUIRED, ENTRY_VARIANT_UID_REQUIRED +from .._messages import ( + ENTRY_BODY_REQUIRED, + ENTRY_VARIANT_CONTENT_TYPE_UID_REQUIRED, + ENTRY_VARIANT_ENTRY_UID_REQUIRED, + ENTRY_VARIANT_UID_REQUIRED, +) class EntryVariants(Parameter): """ This class takes a base URL as an argument when it's initialized, which is the endpoint for the RESTFUL API that - we'll be interacting with. The query(), create(), fetch(), delete(), update(), versions(), and includeVariants() - methods each correspond to the operations that can be performed on the API """ + we'll be interacting with. The query(), create(), fetch(), delete(), update(), versions(), + includeVariants(), publish(), and unpublish() methods each correspond to the operations that can be + performed on the API. Optional ``branch`` scopes requests to a stack branch via the ``branch`` header. """ - def __init__(self, client, content_type_uid: str, entry_uid: str, variant_uid: str = None): + def __init__( + self, + client, + content_type_uid: str, + entry_uid: str, + variant_uid: str = None, + branch: str = None, + ): self.client = client self.content_type_uid = content_type_uid self.entry_uid = entry_uid self.variant_uid = variant_uid + self.branch = branch if branch else None super().__init__(self.client) self.path = f"content_types/{content_type_uid}/entries/{entry_uid}/variants" + def _headers(self): + """Merge optional branch into request headers without mutating the client.""" + if not self.branch: + return self.client.headers + headers = dict(self.client.headers) + headers["branch"] = self.branch + return headers def find(self, params: dict = None): """ @@ -47,7 +68,7 @@ def find(self, params: dict = None): self.validate_entry_uid() if params is not None: self.params.update(params) - return self.client.get(self.path, headers = self.client.headers, params = self.params) + return self.client.get(self.path, headers=self._headers(), params=self.params) def create(self, data: dict): """ @@ -79,7 +100,7 @@ def create(self, data: dict): self.validate_content_type_uid() self.validate_entry_uid() data = json.dumps(data) - return self.client.post(self.path, headers = self.client.headers, data=data, params = self.params) + return self.client.post(self.path, headers=self._headers(), data=data, params=self.params) def fetch(self, variant_uid: str = None, params: dict = None): """ @@ -96,9 +117,11 @@ def fetch(self, variant_uid: str = None, params: dict = None): >>> import contentstack_management >>> client = contentstack_management.Client(authtoken='your_authtoken') - >>> result = client.stack('api_key').content_types('content_type_uid').entry('entry_uid').variants('variant_uid').fetch().json() + >>> result = client.stack('api_key').content_types('content_type_uid').entry('entry_uid').variants().fetch('variant_uid').json() >>> # With parameters - >>> result = client.stack('api_key').content_types('content_type_uid').entry('entry_uid').variants('variant_uid').fetch(params={'include_count': True}).json() + >>> result = client.stack('api_key').content_types('content_type_uid').entry('entry_uid').variants().fetch('variant_uid', params={'include_count': True}).json() + >>> # On a branch + >>> result = client.stack('api_key').content_types('content_type_uid').entry('entry_uid').variants('variant_uid', 'branch_uid').fetch().json() ------------------------------- """ @@ -112,7 +135,7 @@ def fetch(self, variant_uid: str = None, params: dict = None): if params is not None: self.params.update(params) url = f"{self.path}/{self.variant_uid}" - return self.client.get(url, headers = self.client.headers, params = self.params) + return self.client.get(url, headers=self._headers(), params=self.params) def delete(self, variant_uid: str = None): """ @@ -128,7 +151,7 @@ def delete(self, variant_uid: str = None): >>> import contentstack_management >>> client = contentstack_management.Client(authtoken='your_authtoken') - >>> result = client.stack('api_key').content_types('content_type_uid').entry('entry_uid').variants('variant_uid').delete().json() + >>> result = client.stack('api_key').content_types('content_type_uid').entry('entry_uid').variants().delete('variant_uid').json() ------------------------------- """ @@ -138,7 +161,7 @@ def delete(self, variant_uid: str = None): self.validate_entry_uid() self.validate_variant_uid() url = f"{self.path}/{self.variant_uid}" - return self.client.delete(url, headers = self.client.headers, params = self.params) + return self.client.delete(url, headers=self._headers(), params=self.params) def update(self, data: dict, variant_uid: str = None): """ @@ -167,7 +190,7 @@ def update(self, data: dict, variant_uid: str = None): >>> } >>> import contentstack_management >>> client = contentstack_management.Client(authtoken='your_authtoken') - >>> result = client.stack('api_key').content_types('content_type_uid').entry('entry_uid').variants('variant_uid').update(data).json() + >>> result = client.stack('api_key').content_types('content_type_uid').entry('entry_uid').variants().update(data, 'variant_uid').json() ------------------------------- """ @@ -178,7 +201,7 @@ def update(self, data: dict, variant_uid: str = None): self.validate_variant_uid() url = f"{self.path}/{self.variant_uid}" data = json.dumps(data) - return self.client.put(url, headers = self.client.headers, data=data, params = self.params) + return self.client.put(url, headers=self._headers(), data=data, params=self.params) def versions(self, variant_uid: str = None, params: dict = None): """ @@ -195,9 +218,9 @@ def versions(self, variant_uid: str = None, params: dict = None): >>> import contentstack_management >>> client = contentstack_management.Client(authtoken='your_authtoken') - >>> result = client.stack('api_key').content_types('content_type_uid').entry('entry_uid').variants('variant_uid').versions().json() + >>> result = client.stack('api_key').content_types('content_type_uid').entry('entry_uid').variants().versions('variant_uid').json() >>> # With parameters - >>> result = client.stack('api_key').content_types('content_type_uid').entry('entry_uid').variants('variant_uid').versions(params={'limit': 10}).json() + >>> result = client.stack('api_key').content_types('content_type_uid').entry('entry_uid').variants().versions('variant_uid', params={'limit': 10}).json() ------------------------------- """ @@ -209,7 +232,7 @@ def versions(self, variant_uid: str = None, params: dict = None): if params is not None: self.params.update(params) url = f"{self.path}/{self.variant_uid}/versions" - return self.client.get(url, headers = self.client.headers, params = self.params) + return self.client.get(url, headers=self._headers(), params=self.params) def includeVariants(self, include_variants: str = 'true', variant_uid: str = None, params: dict = None): """ @@ -243,8 +266,85 @@ def includeVariants(self, include_variants: str = 'true', variant_uid: str = Non self.params.update(params) self.params['include_variants'] = include_variants url = f"content_types/{self.content_type_uid}/entries/{self.entry_uid}" - return self.client.get(url, headers = self.client.headers, params = self.params) + return self.client.get(url, headers=self._headers(), params=self.params) + def publish(self, data: dict): + """ + Publish the entry for this content type and entry UID (CMA ``.../entries/{entry_uid}/publish``). + + For entry variants, the body typically includes ``entry.environments``, ``entry.locales``, + ``entry.variants`` (list of ``{"uid", "version"}`` per variant), optional ``entry.variant_rules``, + and top-level ``locale`` (and optional scheduling fields supported by the API). + + :param data: Publish payload dict (serialized as JSON). + :return: Response from the publish request. + ------------------------------- + [Example:] + + >>> data = { + >>> "entry": { + >>> "environments": ["production"], + >>> "locales": ["en-us"], + >>> "variants": [ + >>> {"uid": "variant_uid", "version": 1} + >>> ], + >>> "variant_rules": { + >>> "publish_latest_base": False, + >>> "publish_latest_base_conditionally": True + >>> } + >>> }, + >>> "locale": "en-us" + >>> } + >>> import contentstack_management + >>> client = contentstack_management.Client(authtoken='your_authtoken') + >>> result = client.stack('api_key').content_types('content_type_uid').entry('entry_uid').variants().publish(data).json() + + ------------------------------- + """ + self.validate_content_type_uid() + self.validate_entry_uid() + if data is None: + raise Exception(ENTRY_BODY_REQUIRED) + url = f"content_types/{self.content_type_uid}/entries/{self.entry_uid}/publish" + data = json.dumps(data) + return self.client.post(url, headers=self._headers(), data=data, params=self.params) + + def unpublish(self, data: dict): + """ + Unpublish the entry for this content type and entry UID (CMA ``.../entries/{entry_uid}/unpublish``). + + For entry variants, the body typically includes ``entry.environments``, ``entry.locales``, + ``entry.variants`` (list of ``{"uid", "version"}`` per variant), and top-level ``locale``. + + :param data: Unpublish payload dict (serialized as JSON). + :return: Response from the unpublish request. + ------------------------------- + [Example:] + + >>> data = { + >>> "entry": { + >>> "environments": ["environment_uid"], + >>> "locales": ["en-us"], + >>> "variants": [ + >>> {"uid": "variant_uid", "version": 1} + >>> ] + >>> }, + >>> "locale": "en-us" + >>> } + >>> import contentstack_management + >>> client = contentstack_management.Client(authtoken='your_authtoken') + >>> result = client.stack('api_key').content_types('content_type_uid').entry('entry_uid').variants().unpublish(data).json() + + ------------------------------- + """ + self.validate_content_type_uid() + self.validate_entry_uid() + if data is None: + raise Exception(ENTRY_BODY_REQUIRED) + url = f"content_types/{self.content_type_uid}/entries/{self.entry_uid}/unpublish" + data = json.dumps(data) + return self.client.post(url, headers=self._headers(), data=data, params=self.params) + def validate_content_type_uid(self): """ The function checks if the content_type_uid is None or an empty string and raises an ArgumentException diff --git a/tests/unit/entry_variants/test_entry_variants.py b/tests/unit/entry_variants/test_entry_variants.py index 2eabed9..53eea4b 100644 --- a/tests/unit/entry_variants/test_entry_variants.py +++ b/tests/unit/entry_variants/test_entry_variants.py @@ -59,7 +59,7 @@ def test_create_entry_variant(self): self.assertEqual(response.request.headers["Content-Type"], "application/json") def test_fetch_entry_variant(self): - response = self.client.stack(api_key).content_types(content_type_uid).entry(entry_uid).variants(variant_uid).fetch() + response = self.client.stack(api_key).content_types(content_type_uid).entry(entry_uid).variants(variant_uid, None).fetch() self.assertEqual(response.request.url, f"{self.client.endpoint}content_types/{content_type_uid}/entries/{entry_uid}/variants/{variant_uid}") self.assertEqual(response.request.method, "GET") self.assertEqual(response.request.headers["Content-Type"], "application/json") @@ -67,7 +67,7 @@ def test_fetch_entry_variant(self): def test_fetch_entry_variant_with_params(self): params = {"include_count": True} - response = self.client.stack(api_key).content_types(content_type_uid).entry(entry_uid).variants(variant_uid).fetch(params=params) + response = self.client.stack(api_key).content_types(content_type_uid).entry(entry_uid).variants(variant_uid, None).fetch(params=params) self.assertEqual(response.request.url, f"{self.client.endpoint}content_types/{content_type_uid}/entries/{entry_uid}/variants/{variant_uid}?include_count=True") self.assertEqual(response.request.method, "GET") self.assertEqual(response.request.headers["Content-Type"], "application/json") @@ -86,19 +86,19 @@ def test_update_entry_variant(self): "description": "Updated description" } } - response = self.client.stack(api_key).content_types(content_type_uid).entry(entry_uid).variants(variant_uid).update(data) + response = self.client.stack(api_key).content_types(content_type_uid).entry(entry_uid).variants(variant_uid, None).update(data) self.assertEqual(response.request.url, f"{self.client.endpoint}content_types/{content_type_uid}/entries/{entry_uid}/variants/{variant_uid}") self.assertEqual(response.request.method, "PUT") self.assertEqual(response.request.headers["Content-Type"], "application/json") def test_delete_entry_variant(self): - response = self.client.stack(api_key).content_types(content_type_uid).entry(entry_uid).variants(variant_uid).delete() + response = self.client.stack(api_key).content_types(content_type_uid).entry(entry_uid).variants(variant_uid, None).delete() self.assertEqual(response.request.url, f"{self.client.endpoint}content_types/{content_type_uid}/entries/{entry_uid}/variants/{variant_uid}") self.assertEqual(response.request.method, "DELETE") self.assertEqual(response.request.headers["Content-Type"], "application/json") def test_versions_entry_variant(self): - response = self.client.stack(api_key).content_types(content_type_uid).entry(entry_uid).variants(variant_uid).versions() + response = self.client.stack(api_key).content_types(content_type_uid).entry(entry_uid).variants(variant_uid, None).versions() self.assertEqual(response.request.url, f"{self.client.endpoint}content_types/{content_type_uid}/entries/{entry_uid}/variants/{variant_uid}/versions") self.assertEqual(response.request.method, "GET") self.assertEqual(response.request.headers["Content-Type"], "application/json") @@ -106,7 +106,7 @@ def test_versions_entry_variant(self): def test_versions_entry_variant_with_params(self): params = {"limit": 10, "skip": 0} - response = self.client.stack(api_key).content_types(content_type_uid).entry(entry_uid).variants(variant_uid).versions(params=params) + response = self.client.stack(api_key).content_types(content_type_uid).entry(entry_uid).variants(variant_uid, None).versions(params=params) self.assertEqual(response.request.url, f"{self.client.endpoint}content_types/{content_type_uid}/entries/{entry_uid}/variants/{variant_uid}/versions?limit=10&skip=0") self.assertEqual(response.request.method, "GET") self.assertEqual(response.request.headers["Content-Type"], "application/json") @@ -127,6 +127,66 @@ def test_include_variants_with_params(self): self.assertEqual(response.request.headers["Content-Type"], "application/json") self.assertEqual(response.request.body, None) + def test_find_entry_variants_with_branch_header(self): + branch_uid = "main" + response = self.client.stack(api_key).content_types(content_type_uid).entry(entry_uid).variants(branch_uid).find() + self.assertEqual( + response.request.url, + f"{self.client.endpoint}content_types/{content_type_uid}/entries/{entry_uid}/variants", + ) + self.assertEqual(response.request.headers["branch"], branch_uid) + + def test_fetch_entry_variant_with_uid_and_branch(self): + branch_uid = "main" + response = self.client.stack(api_key).content_types(content_type_uid).entry(entry_uid).variants(variant_uid, branch_uid).fetch() + self.assertEqual( + response.request.url, + f"{self.client.endpoint}content_types/{content_type_uid}/entries/{entry_uid}/variants/{variant_uid}", + ) + self.assertEqual(response.request.headers["branch"], branch_uid) + + def test_variants_rejects_more_than_two_positional_args(self): + with self.assertRaises(TypeError): + self.client.stack(api_key).content_types(content_type_uid).entry(entry_uid).variants("a", "b", "c") + + def test_publish_entry_from_variants(self): + data = { + "entry": { + "environments": ["production"], + "locales": ["en-us"], + "variants": [{"uid": variant_uid, "version": 1}], + "variant_rules": { + "publish_latest_base": False, + "publish_latest_base_conditionally": True, + }, + }, + "locale": "en-us", + } + response = self.client.stack(api_key).content_types(content_type_uid).entry(entry_uid).variants().publish(data) + self.assertEqual( + response.request.url, + f"{self.client.endpoint}content_types/{content_type_uid}/entries/{entry_uid}/publish", + ) + self.assertEqual(response.request.method, "POST") + self.assertEqual(response.request.headers["Content-Type"], "application/json") + + def test_unpublish_entry_from_variants(self): + data = { + "entry": { + "environments": ["development"], + "locales": ["en-us"], + "variants": [{"uid": variant_uid, "version": 1}], + }, + "locale": "en-us", + } + response = self.client.stack(api_key).content_types(content_type_uid).entry(entry_uid).variants().unpublish(data) + self.assertEqual( + response.request.url, + f"{self.client.endpoint}content_types/{content_type_uid}/entries/{entry_uid}/unpublish", + ) + self.assertEqual(response.request.method, "POST") + self.assertEqual(response.request.headers["Content-Type"], "application/json") + def test_validate_content_type_uid_with_valid_uid(self): entry_variants = self.client.stack(api_key).content_types(content_type_uid).entry(entry_uid).variants() # This should not raise an exception