Merge pull request #143 from contentstack/staging

DX | 14-07-2025 | Release

Author: harshithad0703

CHANGELOG.md

@@ -1,5 +1,17 @@
 # CHANGELOG
 
+## _v2.2.0_
+
+### **Date: 14-July-2025**
+
+- Variants Support Added.
+
+## _v2.1.1_
+
+### **Date: 07-July-2025**
+
+- Fixed sanity testcases and removed hardcoded secrets.
+
 ## _v2.1.0_
 
 ### **Date: 02-June-2025**

contentstack/__init__.py

@@ -22,7 +22,7 @@
 __title__ = 'contentstack-delivery-python'
 __author__ = 'contentstack'
 __status__ = 'debug'
-__version__ = 'v2.1.0'
+__version__ = 'v2.2.0'
 __endpoint__ = 'cdn.contentstack.io'
 __email__ = '[email protected]'
 __developer_email__ = '[email protected]'

contentstack/contenttype.py

@@ -13,6 +13,7 @@
 
 from contentstack.entry import Entry
 from contentstack.query import Query
+from contentstack.variants import Variants
 
 class ContentType:
     """
@@ -118,3 +119,18 @@ def find(self, params=None):
         url = f'{endpoint}/content_types?{encoded_params}'
         result = self.http_instance.get(url)
         return result
+    
+    def variants(self, variant_uid: str | list[str], params: dict = None):
+        """
+        Fetches the variants of the content type
+        :param variant_uid: {str} -- variant_uid
+        :return: Entry, 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,
+            params=params,
+            logger=None
+        )

contentstack/entry.py

@@ -8,6 +8,7 @@
 
 from contentstack.deep_merge_lp import DeepMergeMixin
 from contentstack.entryqueryable import EntryQueryable
+from contentstack.variants import Variants
 
 class Entry(EntryQueryable):
     """
@@ -222,6 +223,22 @@ def _merged_response(self):
             merged_response = DeepMergeMixin(entry_response, lp_entry).to_dict()  # Convert to dictionary
             return merged_response  # Now correctly returns a dictionary
         raise ValueError("Missing required keys in live_preview data")
+    
+    def variants(self, variant_uid: str | list[str], params: dict = None):
+        """
+        Fetches the variants of the entry
+        :param variant_uid: {str} -- variant_uid
+        :return: Entry, 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,
+            params=params,
+            logger=self.logger
+        )
+
 
 
 

contentstack/variants.py

@@ -0,0 +1,93 @@
+import logging
+from urllib import parse
+
+from contentstack.entryqueryable import EntryQueryable
+
+class Variants(EntryQueryable):
+    """
+    An entry is the actual piece of content that you want to publish.
+    Entries can be created for one of the available content types.
+
+    Entry works with
+    version={version_number}
+    environment={environment_name}
+    locale={locale_code}
+    """
+
+    def __init__(self,
+        http_instance=None,
+        content_type_uid=None,
+        entry_uid=None,
+        variant_uid=None,
+        params=None,
+        logger=None):
+        
+        super().__init__()
+        EntryQueryable.__init__(self)
+        self.entry_param = {}
+        self.http_instance = http_instance
+        self.content_type_id = content_type_uid
+        self.entry_uid = entry_uid
+        self.variant_uid = variant_uid
+        self.logger = logger or logging.getLogger(__name__)
+        self.entry_param = params or {}
+
+    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)
+        
+        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)
+        result = self.http_instance.get(url)
+        self.http_instance.headers.pop('x-cs-variant-uid', None)
+        return result
+    
+    def fetch(self, params=None):
+        """
+        This method is useful to fetch variant entries of a particular content type and entries of the of the stack.
+        :return:dict -- contentType response
+        ------------------------------
+        Example:
+
+            >>> import contentstack
+            >>> stack = contentstack.Stack('api_key', 'delivery_token', 'environment')
+            >>> content_type = stack.content_type('content_type_uid')
+            >>> some_dict = {'abc':'something'}
+            >>> response = content_type.fetch(some_dict)
+        ------------------------------
+        """
+        """
+        Fetches the variants of the entry
+        :param self.variant_uid: {str} -- self.variant_uid
+        :return: Entry, so you can chain this call.
+        """
+        if self.entry_uid is None:
+            raise ValueError("entry_uid is 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)
+            
+            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)
+            result = self.http_instance.get(url)
+            self.http_instance.headers.pop('x-cs-variant-uid', None)
+            return result

requirements.txt

@@ -2,7 +2,7 @@ twython==3.9.1
 setuptools==80.3.1
 contentstack-utils==1.3.0
 python-dateutil==2.8.2
-requests==2.32.3
+requests==2.32.4
 coverage==7.6.0
 tox==4.5.1
 virtualenv==20.26.6
@@ -58,6 +58,6 @@ zipp==3.20.1
 distlib~=0.3.8
 cachetools~=5.4.0
 tomlkit~=0.13.2
-urllib3==2.2.3
+urllib3==2.5.0
 exceptiongroup~=1.2.2
 iniconfig~=2.0.0

tests/test_entry.py

@@ -3,12 +3,12 @@
 import config
 import contentstack
 
-_UID = 'blt53ca1231625bdde4'
 API_KEY = config.APIKEY
 DELIVERY_TOKEN = config.DELIVERYTOKEN
 ENVIRONMENT = config.ENVIRONMENT
 HOST = config.HOST
-
+FAQ_UID = config.FAQ_UID  # Add this in your config.py
+VARIANT_UID = config.VARIANT_UID
 
 class TestEntry(unittest.TestCase):
 
@@ -19,69 +19,54 @@ def test_run_initial_query(self):
         query = self.stack.content_type('faq').query()
         result = query.find()
         if result is not None:
-            self._UID = result['entries'][0]['uid']
-            print(f'the uid is: {_UID}')
+            self.faq_uid = result['entries'][0]['uid']
+            print(f'the uid is: {self.faq_uid}')
 
     def test_entry_by_UID(self):
-        global _UID
-        entry = self.stack.content_type('faq').entry(_UID)
+        entry = self.stack.content_type('faq').entry(FAQ_UID)
         result = entry.fetch()
         if result is not None:
-            _UID = result['entry']['uid']
-            self.assertEqual(_UID, result['entry']['uid'])
+            self.assertEqual(FAQ_UID, result['entry']['uid'])
 
     def test_03_entry_environment(self):
-        global _UID
-        entry = self.stack.content_type('faq').entry(
-            _UID).environment('test')
+        entry = self.stack.content_type('faq').entry(FAQ_UID).environment('test')
         self.assertEqual("test", entry.http_instance.headers['environment'])
 
     def test_04_entry_locale(self):
-        global _UID
-        entry = self.stack.content_type('faq').entry(_UID).locale('en-ei')
+        entry = self.stack.content_type('faq').entry(FAQ_UID).locale('en-ei')
         entry.fetch()
         self.assertEqual('en-ei', entry.entry_param['locale'])
 
     def test_05_entry_version(self):
-        global _UID
-        entry = self.stack.content_type('faq').entry(_UID).version(3)
+        entry = self.stack.content_type('faq').entry(FAQ_UID).version(3)
         entry.fetch()
         self.assertEqual(3, entry.entry_param['version'])
 
     def test_06_entry_params(self):
-        global _UID
-        entry = self.stack.content_type('faq').entry(
-            _UID).param('param_key', 'param_value')
+        entry = self.stack.content_type('faq').entry(FAQ_UID).param('param_key', 'param_value')
         entry.fetch()
         self.assertEqual('param_value', entry.entry_param['param_key'])
 
     def test_07_entry_base_only(self):
-        global _UID
-        entry = self.stack.content_type(
-            'faq').entry(_UID).only('field_UID')
+        entry = self.stack.content_type('faq').entry(FAQ_UID).only('field_UID')
         entry.fetch()
         self.assertEqual({'environment': 'development',
                           'only[BASE][]': 'field_UID'}, entry.entry_param)
 
     def test_08_entry_base_excepts(self):
-        global _UID
-        entry = self.stack.content_type('faq').entry(
-            _UID).excepts('field_UID')
+        entry = self.stack.content_type('faq').entry(FAQ_UID).excepts('field_UID')
         entry.fetch()
         self.assertEqual({'environment': 'development',
                           'except[BASE][]': 'field_UID'}, entry.entry_param)
 
     def test_10_entry_base_include_reference_only(self):
-        global _UID
-        entry = self.stack.content_type('faq').entry(_UID).only('field1')
+        entry = self.stack.content_type('faq').entry(FAQ_UID).only('field1')
         entry.fetch()
         self.assertEqual({'environment': 'development', 'only[BASE][]': 'field1'},
                          entry.entry_param)
 
     def test_11_entry_base_include_reference_excepts(self):
-        global _UID
-        entry = self.stack.content_type(
-            'faq').entry(_UID).excepts('field1')
+        entry = self.stack.content_type('faq').entry(FAQ_UID).excepts('field1')
         entry.fetch()
         self.assertEqual({'environment': 'development', 'except[BASE][]': 'field1'},
                          entry.entry_param)
@@ -95,15 +80,13 @@ def test_12_entry_include_reference_github_issue(self):
         response = _entry.fetch()
 
     def test_13_entry_support_include_fallback_unit_test(self):
-        global _UID
-        entry = self.stack.content_type('faq').entry(
-            _UID).include_fallback()
+        entry = self.stack.content_type('faq').entry(FAQ_UID).include_fallback()
         self.assertEqual(
             True, entry.entry_param.__contains__('include_fallback'))
 
     def test_14_entry_queryable_only(self):
         try:
-            entry = self.stack.content_type('faq').entry(_UID).only(4)
+            entry = self.stack.content_type('faq').entry(FAQ_UID).only(4)
             result = entry.fetch()
             self.assertEqual(None, result['uid'])
         except KeyError as e:
@@ -112,28 +95,25 @@ def test_14_entry_queryable_only(self):
 
     def test_entry_queryable_excepts(self):
         try:
-            entry = self.stack.content_type('faq').entry(_UID).excepts(4)
+            entry = self.stack.content_type('faq').entry(FAQ_UID).excepts(4)
             result = entry.fetch()
             self.assertEqual(None, result['uid'])
         except KeyError as e:
             if hasattr(e, 'message'):
                 self.assertEqual("Invalid field_UID provided", e.args[0])
 
     def test_16_entry_queryable_include_content_type(self):
-        entry = self.stack.content_type('faq').entry(
-            _UID).include_content_type()
+        entry = self.stack.content_type('faq').entry(FAQ_UID).include_content_type()
         self.assertEqual({'include_content_type': 'true', 'include_global_field_schema': 'true'},
                          entry.entry_queryable_param)
 
     def test_reference_content_type_uid(self):
-        entry = self.stack.content_type('faq').entry(
-            _UID).include_reference_content_type_uid()
+        entry = self.stack.content_type('faq').entry(FAQ_UID).include_reference_content_type_uid()
         self.assertEqual({'include_reference_content_type_uid': 'true'},
                          entry.entry_queryable_param)
 
     def test_19_entry_queryable_add_param(self):
-        entry = self.stack.content_type('faq').entry(
-            _UID).add_param('cms', 'contentstack')
+        entry = self.stack.content_type('faq').entry(FAQ_UID).add_param('cms', 'contentstack')
         self.assertEqual({'cms': 'contentstack'}, entry.entry_queryable_param)
 
     def test_20_entry_include_fallback(self):
@@ -154,6 +134,28 @@ def test_22_entry_include_metadata(self):
         content_type = self.stack.content_type('faq')
         entry = content_type.entry("878783238783").include_metadata()
         self.assertEqual({'include_metadata': 'true'}, entry.entry_queryable_param)
+                
+    def test_23_content_type_variants(self):
+        content_type = self.stack.content_type('faq')
+        entry = content_type.variants(VARIANT_UID).find()
+        self.assertIn('variants', entry['entries'][0]['publish_details'])
+        
+    def test_24_entry_variants(self):
+        content_type = self.stack.content_type('faq')
+        entry = content_type.entry(FAQ_UID).variants(VARIANT_UID).fetch()
+        self.assertIn('variants', entry['entry']['publish_details'])
+        
+    def test_25_content_type_variants_with_has_hash_variant(self):
+        content_type = self.stack.content_type('faq')
+        entry = content_type.variants([VARIANT_UID]).find()
+        self.assertIn('variants', entry['entries'][0]['publish_details'])
+        
+    def test_25_content_type_entry_variants_with_has_hash_variant(self):
+        content_type = self.stack.content_type('faq').entry(FAQ_UID)
+        entry = content_type.variants([VARIANT_UID]).fetch()
+        self.assertIn('variants', entry['entry']['publish_details'])
+        
+        
 
 
 if __name__ == '__main__':

tests/test_live_preview.py

@@ -4,9 +4,9 @@
 import contentstack
 from contentstack.deep_merge_lp import DeepMergeMixin
 
-management_token = 'cs8743874323343u9'
-entry_uid = 'blt8743874323343u9'
-preview_token = 'abcdefgh1234567890'
+management_token = config.MANAGEMENT_TOKEN
+entry_uid = config.LIVE_PREVIEW_ENTRY_UID
+preview_token = config.PREVIEW_TOKEN
 
 _lp_query = {
     'live_preview': '#0#0#0#0#0#0#0#0#0#',