Added livepreview 2.0 support and timeline preview support

Author: sunil-lakshman

contentstack/deep_merge_lp.py

@@ -1,21 +1,36 @@
 class DeepMergeMixin:
 
     def __init__(self, entry_response, lp_response):
+        if not isinstance(entry_response, list) or not isinstance(lp_response, list):
+            raise TypeError("Both entry_response and lp_response must be lists of dictionaries")
+
         self.entry_response = entry_response
         self.lp_response = lp_response
+        self.merged_response = self._merge_entries(entry_response, lp_response)
+
+    def _merge_entries(self, entry_list, lp_list):
+        """Merge each LP entry into the corresponding entry response based on UID"""
+        merged_entries = {entry["uid"]: entry.copy() for entry in entry_list}  # Convert to dict for easy lookup
 
-        for lp_obj in self.lp_response:
+        for lp_obj in lp_list:
             uid = lp_obj.get("uid")
-            matching_objs = [entry_obj for entry_obj in entry_response if entry_obj.get("uid") == uid]
-            if matching_objs:
-                for matching_obj in matching_objs:
-                    self._deep_merge(lp_obj, matching_obj)
+            if uid in merged_entries:
+                self._deep_merge(lp_obj, merged_entries[uid])
+            else:
+                merged_entries[uid] = lp_obj  # If LP object does not exist in entry_response, add it
+
+        return list(merged_entries.values())  # Convert back to a list
 
     def _deep_merge(self, source, destination):
+        if not isinstance(destination, dict) or not isinstance(source, dict):
+            return source  # Return source if it's not a dict
         for key, value in source.items():
             if isinstance(value, dict):
                 node = destination.setdefault(key, {})
                 self._deep_merge(value, node)
             else:
                 destination[key] = value
         return destination
+
+    def to_dict(self):
+        return self.merged_response

contentstack/entry.py

@@ -195,7 +195,10 @@ def _impl_live_preview(self):
         if lv is not None and lv['enable'] and 'content_type_uid' in lv and lv[
             'content_type_uid'] == self.content_type_id:
             url = lv['url']
-            self.http_instance.headers['authorization'] = lv['management_token']
+            if lv.get('management_token'):
+                self.http_instance.headers['authorization'] = lv['management_token']
+            else:
+                self.http_instance.headers['preview_token'] = lv['preview_token']
             lp_resp = self.http_instance.get(url)
             if lp_resp is not None and not 'error_code' in lp_resp:
                 self.http_instance.live_preview['lp_response'] = lp_resp
@@ -204,8 +207,27 @@ def _impl_live_preview(self):
 
     def _merged_response(self):
         if 'entry_response' in self.http_instance.live_preview and 'lp_response' in self.http_instance.live_preview:
-            entry_response = self.http_instance.live_preview['entry_response']['entry']
+            entry_response = self.http_instance.live_preview['entry_response']
             lp_response = self.http_instance.live_preview['lp_response']
-            merged_response = DeepMergeMixin(entry_response, lp_response)
-            return merged_response.entry_response
-        pass
+            
+            # Ensure lp_entry exists
+            if 'entry' in lp_response:
+                lp_entry = lp_response['entry']
+            else:
+                raise KeyError(f"'entry' key not found in lp_response: {lp_response}")
+            # 🛠️ Fix: Ensure both are lists of dictionaries
+            if not isinstance(entry_response, list):
+                entry_response = [entry_response]  # Wrap in a list if it's a dict
+            if not isinstance(lp_entry, list):
+                lp_entry = [lp_entry]  # Wrap in a list if it's a dict
+            if not all(isinstance(item, dict) for item in entry_response):
+                raise TypeError(f"entry_response must be a list of dictionaries. Got: {entry_response}")
+            if not all(isinstance(item, dict) for item in lp_entry):
+                raise TypeError(f"lp_entry must be a list of dictionaries. Got: {lp_entry}")
+            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")
+
+
+
+

contentstack/query.py

@@ -321,31 +321,52 @@ def __execute_network_call(self):
             self.query_params["query"] = json.dumps(self.parameters)
         if 'environment' in self.http_instance.headers:
             self.query_params['environment'] = self.http_instance.headers['environment']
+
         encoded_string = parse.urlencode(self.query_params, doseq=True)
         url = f'{self.base_url}?{encoded_string}'
         self._impl_live_preview()
         response = self.http_instance.get(url)
-        if self.http_instance.live_preview is not None and not 'errors' in response:
-            self.http_instance.live_preview['entry_response'] = response['entries']
+        # Ensure response is converted to dictionary
+        if isinstance(response, str):
+            try:
+                response = json.loads(response)  # Convert JSON string to dictionary
+            except json.JSONDecodeError as e:
+                print(f"JSON decode error: {e}")
+                return {"error": "Invalid JSON response"}  # Return an error dictionary
+
+        if self.http_instance.live_preview is not None and 'errors' not in response:
+            if 'entries' in response:
+                self.http_instance.live_preview['entry_response'] = response['entries'][0]  # Get first entry
+            else:
+                print(f"Error: 'entries' key missing in response: {response}")
+                return {"error": "'entries' key missing in response"}
             return self._merged_response()
         return response
 
     def _impl_live_preview(self):
         lv = self.http_instance.live_preview
-        if lv is not None and lv['enable'] and 'content_type_uid' in lv and lv[
-            'content_type_uid'] == self.content_type_uid:
+        if lv is not None and lv.get('enable') and lv.get('content_type_uid') == self.content_type_uid:
             url = lv['url']
-            self.http_instance.headers['authorization'] = lv['management_token']
+            if lv.get('management_token'):
+                self.http_instance.headers['authorization'] = lv['management_token']
+            else:
+                self.http_instance.headers['preview_token'] = lv['preview_token']
             lp_resp = self.http_instance.get(url)
-            if lp_resp is not None and not 'error_code' in lp_resp:
-                self.http_instance.live_preview['lp_response'] = lp_resp
+
+            if lp_resp and 'error_code' not in lp_resp:
+                if 'entry' in lp_resp:
+                    self.http_instance.live_preview['lp_response'] = lp_resp['entry']  # Extract entry
+                else:
+                    print(f"Warning: Missing 'entry' key in lp_response: {lp_resp}")
             return None
         return None
 
     def _merged_response(self):
-        if 'entry_response' in self.http_instance.live_preview and 'lp_response' in self.http_instance.live_preview:
-            entry_response = self.http_instance.live_preview['entry_response']['entries']
-            lp_response = self.http_instance.live_preview['lp_response']
+        live_preview = self.http_instance.live_preview
+        if 'entry_response' in live_preview and 'lp_response' in live_preview:
+            entry_response = live_preview['entry_response']
+            lp_response = live_preview['lp_response']
             merged_response = DeepMergeMixin(entry_response, lp_response)
-            return merged_response.entry_response
-        pass
+            return merged_response  # Return the merged dictionary
+
+        raise ValueError("Missing required keys in live_preview data")

contentstack/stack.py

@@ -41,7 +41,7 @@ def __init__(self, api_key: str, delivery_token: str, environment: str,
                      total=5, backoff_factor=0, status_forcelist=[408, 429]),
                  live_preview=None,
                  branch=None,
-                 early_access =  None,
+                 early_access = None,
                  ):
         """
         # Class that wraps the credentials of the authenticated user. Think of
@@ -96,6 +96,15 @@ def __init__(self, api_key: str, delivery_token: str, environment: str,
         self.live_preview = live_preview
         self.early_access = early_access
         self._validate_stack()
+        self._setup_headers()
+        self._setup_live_preview()
+        self.http_instance = HTTPSConnection(
+            endpoint=self.endpoint,
+            headers=self.headers,
+            timeout=self.timeout,
+            retry_strategy=self.retry_strategy,
+            live_preview=self.live_preview
+        )
 
     def _validate_stack(self):
         if self.api_key is None or self.api_key == '':
@@ -123,6 +132,7 @@ def _validate_stack(self):
             self.host = f'{self.region.value}-{DEFAULT_HOST}'
         self.endpoint = f'https://{self.host}/{self.version}'
 
+    def _setup_headers(self):
         self.headers = {
             'api_key': self.api_key,
             'access_token': self.delivery_token,
@@ -131,18 +141,10 @@ def _validate_stack(self):
         if self.early_access is not None:
             early_access_str = ', '.join(self.early_access)
             self.headers['x-header-ea'] = early_access_str
-
+ 
         if self.branch is not None:
             self.headers['branch'] = self.branch
-
-        self.http_instance = HTTPSConnection(
-            endpoint=self.endpoint,
-            headers=self.headers,
-            timeout=self.timeout,
-            retry_strategy=self.retry_strategy,
-            live_preview=self.live_preview
-        )
-
+   
     @property
     def get_api_key(self):
         """
@@ -323,8 +325,7 @@ def __sync_request(self):
         base_url = f'{self.http_instance.endpoint}/stacks/sync'
         self.sync_param['environment'] = self.http_instance.headers['environment']
         query = parse.urlencode(self.sync_param)
-        url = f'{base_url}?{query}'
-        return self.http_instance.get(url)
+        return self.http_instance.get(f'{base_url}?{query}')
 
     def image_transform(self, image_url, **kwargs):
         """
@@ -341,6 +342,15 @@ def image_transform(self, image_url, **kwargs):
             raise PermissionError(
                 'image_url required for the image_transformation')
         return ImageTransform(self.http_instance, image_url, **kwargs)
+    
+    def _setup_live_preview(self):
+        if self.live_preview and self.live_preview.get("enable"):
+            region_prefix = "" if self.region.value == "us" else f"{self.region.value}-"
+            self.live_preview["host"] = f"{region_prefix}rest-preview.contentstack.com"
+
+            if self.live_preview.get("preview_token"):
+                self.headers["preview_token"] = self.live_preview["preview_token"]
+
 
     def live_preview_query(self, **kwargs):
         """
@@ -361,28 +371,31 @@ def live_preview_query(self, **kwargs):
                 'authorization': 'management_token'
                 )
         """
-
-        if self.live_preview is not None and self.live_preview['enable'] and 'live_preview_query' in kwargs:
-            self.live_preview.update(**kwargs['live_preview_query'])
-            query = kwargs['live_preview_query']
-            if query is not None:
-                self.live_preview['live_preview'] = query['live_preview']
-            else:
-                self.live_preview['live_preview'] = 'init'
-            if 'content_type_uid' in self.live_preview and self.live_preview['content_type_uid'] is not None:
-                self.live_preview['content_type_uid'] = query['content_type_uid']
-            if 'entry_uid' in self.live_preview and self.live_preview['entry_uid'] is not None:
-                self.live_preview['entry_uid'] = query['entry_uid']
-            self._cal_url()
+        if self.live_preview and self.live_preview.get("enable") and "live_preview_query" in kwargs:
+            query = kwargs["live_preview_query"]
+            if isinstance(query, dict):
+                self.live_preview.update(query)
+                self.live_preview["live_preview"] = query.get("live_preview", "init")
+                if "content_type_uid" in query:
+                    self.live_preview["content_type_uid"] = query["content_type_uid"]
+                if "entry_uid" in query:
+                    self.live_preview["entry_uid"] = query["entry_uid"]
+
+                for key in ["release_id", "preview_timestamp"]:
+                    if key in query:
+                        self.http_instance.headers[key] = query[key]
+                    else:
+                        self.http_instance.headers.pop(key, None)
+
+                self._cal_url()
         return self
 
     def _cal_url(self):
-        host = self.live_preview['host']
-        ct = self.live_preview['content_type_uid']
-        url = f'https://{host}/v3/content_types/{ct}/entries'
-        if 'entry_uid' in self.live_preview:
-            uid = self.live_preview['entry_uid']
-            lv = self.live_preview['live_preview']
-            url = f'{url}/{uid}?live_preview={lv}'
-        self.live_preview['url'] = url
-        pass
+        host = self.live_preview.get("host", DEFAULT_HOST)
+        content_type = self.live_preview.get("content_type_uid", "default_content_type")
+        url = f"https://{host}/v3/content_types/{content_type}/entries"
+        entry_uid = self.live_preview.get("entry_uid")
+        live_preview = self.live_preview.get("live_preview", "init")
+        if entry_uid:
+            url = f"{url}/{entry_uid}?live_preview={live_preview}"
+        self.live_preview["url"] = url

tests/__init__.py

@@ -12,16 +12,19 @@
 from .test_entry import TestEntry
 from .test_query import TestQuery
 from .test_stack import TestStack
+from .test_live_preview import TestLivePreviewConfig
 
 
 def all_tests():
     test_module_stack = TestLoader().loadTestsFromTestCase(TestStack)
     test_module_asset = TestLoader().loadTestsFromTestCase(TestAsset)
     test_module_entry = TestLoader().loadTestsFromTestCase(TestEntry)
     test_module_query = TestLoader().loadTestsFromTestCase(TestQuery)
+    test_module_live_preview = TestLoader().loadTestsFromTestCase(TestLivePreviewConfig)
     TestSuite([
         test_module_stack,
         test_module_asset,
         test_module_entry,
         test_module_query,
+        test_module_live_preview
     ])

tests/test_live_preview.py

@@ -6,25 +6,38 @@
 
 management_token = 'cs8743874323343u9'
 entry_uid = 'blt8743874323343u9'
+preview_token = 'abcdefgh1234567890'
 
 _lp_query = {
     'live_preview': '#0#0#0#0#0#0#0#0#0#',
     'content_type_uid': 'product',
     'entry_uid': entry_uid
 }
+_lp_preview_timestamp_query = {
+    'live_preview': '#0#0#0#0#0#0#0#0#0#',
+    'content_type_uid': 'product',
+    'entry_uid': entry_uid,
+    'preview_timestamp': '2025-03-07T12:00:00Z',
+    'release_id': '123456789'
+}
 _lp = {
     'enable': True,
     'host': 'api.contentstack.io',
     'management_token': management_token
 }
 
+_lp_2_0 = {
+    'enable': True,
+    'preview_token': preview_token,
+    'host': 'rest-preview.contentstack.com'
+}
+
 API_KEY = config.APIKEY
 DELIVERY_TOKEN = config.DELIVERYTOKEN
 ENVIRONMENT = config.ENVIRONMENT
 HOST = config.HOST
 ENTRY_UID = config.APIKEY
 
-
 class TestLivePreviewConfig(unittest.TestCase):
 
     def setUp(self):
@@ -58,6 +71,30 @@ def test_021_live_preview_enabled(self):
         self.assertEqual(7, len(self.stack.live_preview))
         self.assertEqual('product', self.stack.live_preview['content_type_uid'])
 
+    def test_022_preview_timestamp_with_livepreview_2_0_enabled(self):
+        self.stack = contentstack.Stack(
+            API_KEY,
+            DELIVERY_TOKEN,
+            ENVIRONMENT,
+            live_preview=_lp_2_0)
+        self.stack.live_preview_query(live_preview_query=_lp_preview_timestamp_query)
+        self.assertIsNotNone(self.stack.live_preview['preview_token'])
+        self.assertEqual(9, len(self.stack.live_preview))
+        self.assertEqual('product', self.stack.live_preview['content_type_uid'])
+        self.assertEqual('123456789', self.stack.live_preview['release_id'])
+        self.assertEqual('2025-03-07T12:00:00Z', self.stack.live_preview['preview_timestamp'])
+
+    def test_023_livepreview_2_0_enabled(self):
+        self.stack = contentstack.Stack(
+            API_KEY,
+            DELIVERY_TOKEN,
+            ENVIRONMENT,
+            live_preview=_lp_2_0)
+        self.stack.live_preview_query(live_preview_query=_lp_query)
+        self.assertIsNotNone(self.stack.live_preview['preview_token'])
+        self.assertEqual(9, len(self.stack.live_preview))
+        self.assertEqual('product', self.stack.live_preview['content_type_uid'])
+
     def test_03_set_host(self):
         self.stack = contentstack.Stack(
             API_KEY,
@@ -205,10 +242,9 @@ def test_setup_live_preview(self):
         self.assertTrue(self.stack.get_live_preview['management_token'])
 
     def test_deep_merge_object(self):
-        merged_response = DeepMergeMixin(entry_response, lp_response)
-        self.assertTrue(isinstance(merged_response.entry_response, list))
-        self.assertEqual(3, len(merged_response.entry_response))
-        print(merged_response.entry_response)
+        merged_response = DeepMergeMixin(entry_response, lp_response).to_dict()
+        self.assertTrue(isinstance(merged_response, list), "Merged response should be a list")
+        self.assertTrue(all(isinstance(entry, dict) for entry in merged_response), "Each item in merged_response should be a dictionary")
 
 
 if __name__ == '__main__':