Merge pull request #199 from contentstack/enh/dx-7324

Added Branches support in entry variants

Author: OMpawar-21

CHANGELOG.md

@@ -1,5 +1,11 @@
 # CHANGELOG
 
+## _v2.7.0_
+
+### **Date: 20-July-2026**
+
+- Added optional branch support for entry variants on `Entry.variants()` and `ContentType.variants()`.
+
 ## _v2.6.1_
 
 ### **Date: 13-July-2026**

contentstack/__init__.py

@@ -42,7 +42,7 @@ def get_contentstack_endpoint(region='us', service='', omit_https=False):
 __title__ = 'contentstack-delivery-python'
 __author__ = 'contentstack'
 __status__ = 'debug'
-__version__ = 'v2.6.1'
+__version__ = 'v2.7.0'
 __endpoint__ = 'cdn.contentstack.io'
 __email__ = 'support@contentstack.com'
 __developer_email__ = 'mobile@contentstack.com'

contentstack/contenttype.py

@@ -119,17 +119,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

@@ -267,17 +267,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,48 @@ 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'])
+
+    @unittest.skipIf(BRANCH is None, "config.BRANCH is required for branch variant tests")
+    def test_41e_content_type_variants_branch_as_keyword_arg(self):
+        """Test content_type.variants() with branch passed as keyword argument"""
+        content_type = self.stack.content_type('faq')
+        result = content_type.variants(VARIANT_UID, branch=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_41f_content_type_variants_branch_with_params(self):
+        """Test content_type.variants() with branch and extra query params"""
+        content_type = self.stack.content_type('faq')
+        result = content_type.variants(VARIANT_UID, branch=BRANCH, params={'locale': 'en-us'}).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')