Added livepreview 2.0 and timeline support (#95)

.github/workflows/jira.yml

@@ -21,7 +21,7 @@ jobs:
           project: ${{ secrets.JIRA_PROJECT }}
           issuetype: ${{ secrets.JIRA_ISSUE_TYPE }}
           summary: |
-            ${{ github.event.pull_request.title }}
+            Snyk | Vulnerability | ${{ github.event.repository.name }} | ${{ github.event.pull_request.title }}
           description: |
             PR: ${{ github.event.pull_request.html_url }}
 

.github/workflows/sca-scan.yml

@@ -11,6 +11,5 @@ jobs:
         uses: snyk/actions/python@master
         env:
           SNYK_TOKEN: ${{ secrets.SNYK_TOKEN }}
-          SNYK_INTEGRATION_VERSION: python-3.11
         with:
           args: --fail-on=all --all-projects --skip-unresolved

CHANGELOG.md

@@ -1,5 +1,10 @@
 # CHANGELOG
 
+## _v1.11.0_
+
+### **Date: 17-MARCH-2025**
+
+- Added Livepreview 2.0 and Timeline support
 
 ## _v1.10.0_
 

CODEOWNERS

@@ -1 +1 @@
-* @contentstack/security-admin
+* @contentstack/security-admin

LICENSE

@@ -1,6 +1,6 @@
 MIT License
 
-Copyright (c) 2012 - 2024 Contentstack. All rights reserved.
+Copyright (c) 2012 - 2025 Contentstack. All rights reserved.
 
 Permission is hereby granted, free of charge, to any person obtaining a copy
 of this software and associated documentation files (the "Software"), to deal

contentstack/__init__.py

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

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,24 @@ 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']
-            lp_response = self.http_instance.live_preview['lp_response']
-            merged_response = DeepMergeMixin(entry_response, lp_response)
-            return merged_response.entry_response
-        pass
+            entry_response = self.http_instance.live_preview['entry_response']
+            # Ensure lp_entry exists
+            if 'entry' in self.http_instance.live_preview.get('lp_response', {}):
+                lp_entry = self.http_instance.live_preview['lp_response']['entry']
+            else:
+                lp_entry = {} 
+            if not isinstance(entry_response, list):
+                entry_response = [entry_response]
+            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'] = {'entry': 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

requirements.txt

@@ -5,13 +5,13 @@ python-dateutil==2.8.2
 requests==2.32.2
 coverage==7.2.6
 tox==4.5.1
-virtualenv==20.23.0
+virtualenv==20.26.6
 Sphinx==7.0.1
 sphinxcontrib-websupport==1.2.4
 pip==23.3.1
 build==0.10.0
-wheel==0.40.0
-lxml==4.9.2
+wheel==0.38.0
+lxml==5.3.1
 utils~=1.0.1
 keyring==23.13.1
 docutils==0.20.1
@@ -33,7 +33,7 @@ pytz==2023.3
 Babel==2.12.1
 pep517==0.13.0
 tomli~=2.0.1
-Werkzeug==3.0.3
+Werkzeug==3.0.6
 Flask~=2.3.2
 click~=8.1.3
 MarkupSafe==2.1.2
@@ -44,7 +44,7 @@ pkginfo==1.9.6
 pylint==3.1.0
 astroid==3.1.0
 mccabe==0.7.0
-platformdirs==3.5.1
+platformdirs==3.9.1
 imagesize==1.4.1
 snowballstemmer~=2.2.0
 Pygments~=2.15.1

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
     ])