Added Brancges support in entry variants

Author: sunil-lakshman

CHANGELOG.md

@@ -1,5 +1,11 @@
 # CHANGELOG
 
+## _v2.6.0_
+
+### **Date: 18-May-2026**
+
+- Added optional branch support for entry variants on `Entry.variants()` and `ContentType.variants()`.
+
 ## _v2.5.1_
 
 ### **Date: 15-April-2026**

contentstack/__init__.py

@@ -22,7 +22,7 @@
 __title__ = 'contentstack-delivery-python'
 __author__ = 'contentstack'
 __status__ = 'debug'
-__version__ = 'v2.5.1'
+__version__ = 'v2.6.0'
 __endpoint__ = 'cdn.contentstack.io'
 __email__ = 'support@contentstack.com'
 __developer_email__ = 'mobile@contentstack.com'

contentstack/contenttype.py

@@ -118,17 +118,20 @@ def find(self, params=None):
         result = self.http_instance.get(url)
         return result
 
-    def variants(self, variant_uid: str | list[str], params: dict = None):
+    def variants(self, variant_uid: str | list[str], branch: str = None, params: dict = None):
         """
         Fetches the variants of the content type
-        :param variant_uid: {str} -- variant_uid
-        :return: Entry, so you can chain this call.
+        :param variant_uid: {str | list[str]} -- variant UID or list of variant UIDs
+        :param branch: {str} -- optional branch name to scope the variant request
+        :param params: {dict} -- optional query parameters
+        :return: Variants, so you can chain this call.
         """
         return Variants(
             http_instance=self.http_instance,
             content_type_uid=self.__content_type_uid,
             entry_uid=None,
             variant_uid=variant_uid,
+            branch=branch,
             params=params,
             logger=None
         )

contentstack/entry.py

@@ -266,17 +266,20 @@ def _merged_response(self):
             return merged_response  # Now correctly returns a dictionary
         raise ValueError(ErrorMessages.MISSING_LIVE_PREVIEW_KEYS)
 
-    def variants(self, variant_uid: str | list[str], params: dict = None):
+    def variants(self, variant_uid: str | list[str], branch: str = None, params: dict = None):
         """
         Fetches the variants of the entry
-        :param variant_uid: {str} -- variant_uid
-        :return: Entry, so you can chain this call.
+        :param variant_uid: {str | list[str]} -- variant UID or list of variant UIDs
+        :param branch: {str} -- optional branch name to scope the variant request
+        :param params: {dict} -- optional query parameters
+        :return: Variants, so you can chain this call.
         """
         return Variants(
             http_instance=self.http_instance,
             content_type_uid=self.content_type_id,
             entry_uid=self.entry_uid,
             variant_uid=variant_uid,
+            branch=branch,
             params=params,
             logger=self.logger
         )

contentstack/variants.py

@@ -20,6 +20,7 @@ def __init__(self,
         content_type_uid=None,
         entry_uid=None,
         variant_uid=None,
+        branch=None,
         params=None,
         logger=None):
 
@@ -30,29 +31,47 @@ def __init__(self,
         self.content_type_id = content_type_uid
         self.entry_uid = entry_uid
         self.variant_uid = variant_uid
+        self.branch = branch
         self.logger = logger or logging.getLogger(__name__)
         self.entry_param = params or {}
 
+    def _prepare_variant_headers(self):
+        headers = self.http_instance.headers.copy()
+        if isinstance(self.variant_uid, str):
+            headers['x-cs-variant-uid'] = self.variant_uid
+        elif isinstance(self.variant_uid, list):
+            headers['x-cs-variant-uid'] = ','.join(self.variant_uid)
+        if self.branch is not None:
+            headers['branch'] = self.branch
+        return headers
+
+    def _apply_variant_headers(self, headers):
+        self._original_branch = self.http_instance.headers.get('branch')
+        self.http_instance.headers.update(headers)
+
+    def _cleanup_variant_headers(self):
+        self.http_instance.headers.pop('x-cs-variant-uid', None)
+        if self.branch is not None:
+            if self._original_branch is not None:
+                self.http_instance.headers['branch'] = self._original_branch
+            else:
+                self.http_instance.headers.pop('branch', None)
+
     def find(self, params=None):
         """
         find the variants of the entry of a particular content type
         :param self.variant_uid: {str} -- self.variant_uid
         :return: Entry, so you can chain this call.
         """
-        headers = self.http_instance.headers.copy()  # Create a local copy of headers
-        if isinstance(self.variant_uid, str):
-            headers['x-cs-variant-uid'] = self.variant_uid
-        elif isinstance(self.variant_uid, list):
-            headers['x-cs-variant-uid'] = ','.join(self.variant_uid)
-        
+        headers = self._prepare_variant_headers()
         if params is not None:
             self.entry_param.update(params)
         encoded_params = parse.urlencode(self.entry_param)
         endpoint = self.http_instance.endpoint
         url = f'{endpoint}/content_types/{self.content_type_id}/entries?{encoded_params}'
-        self.http_instance.headers.update(headers)
+        self._apply_variant_headers(headers)
         result = self.http_instance.get(url)
-        self.http_instance.headers.pop('x-cs-variant-uid', None)
+        self._cleanup_variant_headers()
         return result
 
     def fetch(self, params=None):
@@ -77,18 +96,13 @@ def fetch(self, params=None):
         if self.entry_uid is None:
             raise ValueError(ErrorMessages.ENTRY_UID_REQUIRED)
         else:
-            headers = self.http_instance.headers.copy()  # Create a local copy of headers
-            if isinstance(self.variant_uid, str):
-                headers['x-cs-variant-uid'] = self.variant_uid
-            elif isinstance(self.variant_uid, list):
-                headers['x-cs-variant-uid'] = ','.join(self.variant_uid)
-            
+            headers = self._prepare_variant_headers()
             if params is not None:
                 self.entry_param.update(params)
             encoded_params = parse.urlencode(self.entry_param)
             endpoint = self.http_instance.endpoint
             url = f'{endpoint}/content_types/{self.content_type_id}/entries/{self.entry_uid}?{encoded_params}'
-            self.http_instance.headers.update(headers)
+            self._apply_variant_headers(headers)
             result = self.http_instance.get(url)
-            self.http_instance.headers.pop('x-cs-variant-uid', None)
+            self._cleanup_variant_headers()
             return result

tests/test_entry.py

@@ -9,6 +9,7 @@
 HOST = config.HOST
 FAQ_UID = config.FAQ_UID  # Add this in your config.py
 VARIANT_UID = config.VARIANT_UID
+BRANCH = getattr(config, 'BRANCH', None)
 
 class TestEntry(unittest.TestCase):
 
@@ -318,6 +319,34 @@ def test_41_entry_variants_multiple_uids(self):
         result = entry.fetch()
         self.assertIn('variants', result['entry']['publish_details'])
 
+    @unittest.skipIf(BRANCH is None, "config.BRANCH is required for branch variant tests")
+    def test_41a_entry_variants_with_branch(self):
+        """Test single entry variants with branch"""
+        content_type = self.stack.content_type('faq')
+        result = content_type.entry(FAQ_UID).variants(VARIANT_UID, BRANCH).fetch()
+        self.assertIn('variants', result['entry']['publish_details'])
+
+    @unittest.skipIf(BRANCH is None, "config.BRANCH is required for branch variant tests")
+    def test_41b_entry_variants_multiple_uids_with_branch(self):
+        """Test single entry variants with multiple UIDs and branch"""
+        content_type = self.stack.content_type('faq')
+        result = content_type.entry(FAQ_UID).variants([VARIANT_UID], BRANCH).fetch()
+        self.assertIn('variants', result['entry']['publish_details'])
+
+    @unittest.skipIf(BRANCH is None, "config.BRANCH is required for branch variant tests")
+    def test_41c_content_type_variants_find_with_branch(self):
+        """Test entries query variants with branch"""
+        content_type = self.stack.content_type('faq')
+        result = content_type.variants(VARIANT_UID, BRANCH).find()
+        self.assertIn('variants', result['entries'][0]['publish_details'])
+
+    @unittest.skipIf(BRANCH is None, "config.BRANCH is required for branch variant tests")
+    def test_41d_content_type_variants_multiple_uids_find_with_branch(self):
+        """Test entries query variants with multiple UIDs and branch"""
+        content_type = self.stack.content_type('faq')
+        result = content_type.variants([VARIANT_UID], BRANCH).find()
+        self.assertIn('variants', result['entries'][0]['publish_details'])
+
     def test_42_entry_environment_removal(self):
         """Test entry remove_environment method"""
         entry = (self.stack.content_type('faq')

tests/test_variants.py

@@ -0,0 +1,190 @@
+"""
+Unit tests for Variants branch support in contentstack.variants
+"""
+
+import pytest
+from unittest.mock import MagicMock
+from urllib.parse import urlencode
+
+from contentstack.variants import Variants
+
+
+@pytest.fixture
+def mock_http_instance():
+    mock = MagicMock()
+    mock.endpoint = "https://cdn.contentstack.io/v3"
+    mock.headers = {
+        "api_key": "api_key",
+        "access_token": "delivery_token",
+        "environment": "test_env",
+    }
+    mock.get = MagicMock(return_value={"entries": []})
+    return mock
+
+
+@pytest.fixture
+def variants(mock_http_instance):
+    return Variants(
+        http_instance=mock_http_instance,
+        content_type_uid="faq",
+        entry_uid="entry_uid",
+        variant_uid="variant_uid",
+    )
+
+
+def _capture_headers_on_get(mock_http_instance):
+    captured = {}
+
+    def _get(url):
+        captured["headers"] = mock_http_instance.headers.copy()
+        return {"entries": []}
+
+    mock_http_instance.get.side_effect = _get
+    return captured
+
+
+class TestVariantsBranch:
+    def test_fetch_sets_variant_and_branch_headers(self, mock_http_instance):
+        captured = _capture_headers_on_get(mock_http_instance)
+        variants = Variants(
+            http_instance=mock_http_instance,
+            content_type_uid="faq",
+            entry_uid="entry_uid",
+            variant_uid="variant_uid",
+            branch="dev_branch",
+        )
+        variants.fetch()
+
+        assert captured["headers"]["x-cs-variant-uid"] == "variant_uid"
+        assert captured["headers"]["branch"] == "dev_branch"
+
+    def test_fetch_multiple_variant_uids_with_branch(self, mock_http_instance):
+        captured = _capture_headers_on_get(mock_http_instance)
+        variants = Variants(
+            http_instance=mock_http_instance,
+            content_type_uid="faq",
+            entry_uid="entry_uid",
+            variant_uid=["variant1", "variant2"],
+            branch="dev_branch",
+        )
+        variants.fetch()
+
+        assert captured["headers"]["x-cs-variant-uid"] == "variant1,variant2"
+        assert captured["headers"]["branch"] == "dev_branch"
+
+    def test_find_sets_variant_and_branch_headers(self, mock_http_instance):
+        captured = _capture_headers_on_get(mock_http_instance)
+        variants = Variants(
+            http_instance=mock_http_instance,
+            content_type_uid="faq",
+            entry_uid=None,
+            variant_uid="variant_uid",
+            branch="dev_branch",
+        )
+        variants.find()
+
+        assert captured["headers"]["x-cs-variant-uid"] == "variant_uid"
+        assert captured["headers"]["branch"] == "dev_branch"
+
+    def test_fetch_restores_stack_branch_after_request(self, mock_http_instance):
+        mock_http_instance.headers["branch"] = "main"
+        variants = Variants(
+            http_instance=mock_http_instance,
+            content_type_uid="faq",
+            entry_uid="entry_uid",
+            variant_uid="variant_uid",
+            branch="dev_branch",
+        )
+        variants.fetch()
+
+        assert "x-cs-variant-uid" not in mock_http_instance.headers
+        assert mock_http_instance.headers["branch"] == "main"
+
+    def test_fetch_removes_branch_when_stack_had_none(self, mock_http_instance):
+        variants = Variants(
+            http_instance=mock_http_instance,
+            content_type_uid="faq",
+            entry_uid="entry_uid",
+            variant_uid="variant_uid",
+            branch="dev_branch",
+        )
+        variants.fetch()
+
+        assert "branch" not in mock_http_instance.headers
+        assert "x-cs-variant-uid" not in mock_http_instance.headers
+
+    def test_fetch_without_branch_uses_stack_branch(self, mock_http_instance):
+        mock_http_instance.headers["branch"] = "main"
+        variants = Variants(
+            http_instance=mock_http_instance,
+            content_type_uid="faq",
+            entry_uid="entry_uid",
+            variant_uid="variant_uid",
+        )
+        variants.fetch()
+
+        assert mock_http_instance.headers["branch"] == "main"
+        assert "x-cs-variant-uid" not in mock_http_instance.headers
+
+    def test_fetch_cleans_up_variant_header_only(self, variants, mock_http_instance):
+        variants.fetch()
+
+        assert "x-cs-variant-uid" not in mock_http_instance.headers
+        assert mock_http_instance.headers["environment"] == "test_env"
+
+    def test_fetch_builds_expected_url(self, variants, mock_http_instance):
+        variants.fetch()
+        expected_url = (
+            "https://cdn.contentstack.io/v3/content_types/faq/entries/entry_uid?"
+        )
+        mock_http_instance.get.assert_called_once()
+        assert mock_http_instance.get.call_args[0][0].startswith(expected_url)
+
+    def test_find_builds_expected_url(self, mock_http_instance):
+        variants = Variants(
+            http_instance=mock_http_instance,
+            content_type_uid="faq",
+            entry_uid=None,
+            variant_uid="variant_uid",
+            branch="dev_branch",
+        )
+        variants.find(params={"locale": "en-us"})
+        expected_params = urlencode({"locale": "en-us"})
+        expected_url = (
+            f"https://cdn.contentstack.io/v3/content_types/faq/entries?{expected_params}"
+        )
+        mock_http_instance.get.assert_called_once_with(expected_url)
+
+    def test_entry_variants_passes_branch(self, mock_http_instance):
+        from contentstack.entry import Entry
+
+        entry = Entry(
+            http_instance=mock_http_instance,
+            content_type_uid="faq",
+            entry_uid="entry_uid",
+        )
+        result = entry.variants("variant_uid", "dev_branch")
+        assert isinstance(result, Variants)
+        assert result.branch == "dev_branch"
+        assert result.variant_uid == "variant_uid"
+
+    def test_content_type_variants_passes_branch(self, mock_http_instance):
+        from contentstack.contenttype import ContentType
+
+        content_type = ContentType(mock_http_instance, "faq")
+        result = content_type.variants(["variant1", "variant2"], "dev_branch")
+        assert isinstance(result, Variants)
+        assert result.branch == "dev_branch"
+        assert result.variant_uid == ["variant1", "variant2"]
+
+    def test_variants_backward_compatible_params_kwarg(self, mock_http_instance):
+        from contentstack.entry import Entry
+
+        entry = Entry(
+            http_instance=mock_http_instance,
+            content_type_uid="faq",
+            entry_uid="entry_uid",
+        )
+        result = entry.variants("variant_uid", params={"locale": "en-us"})
+        assert result.branch is None
+        assert result.entry_param == {"locale": "en-us"}