diff --git a/.github/workflows/jira.yml b/.github/workflows/jira.yml index caa4bbd..250abc7 100644 --- a/.github/workflows/jira.yml +++ b/.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 }} diff --git a/.github/workflows/sca-scan.yml b/.github/workflows/sca-scan.yml index 18535bb..54d7231 100644 --- a/.github/workflows/sca-scan.yml +++ b/.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 diff --git a/CHANGELOG.md b/CHANGELOG.md index d372fbd..0c91664 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,5 +1,10 @@ # CHANGELOG +## _v1.11.0_ + +### **Date: 17-MARCH-2025** + +- Added Livepreview 2.0 and Timeline support ## _v1.10.0_ diff --git a/CODEOWNERS b/CODEOWNERS index 0773923..1be7e0d 100644 --- a/CODEOWNERS +++ b/CODEOWNERS @@ -1 +1 @@ -* @contentstack/security-admin \ No newline at end of file +* @contentstack/security-admin diff --git a/LICENSE b/LICENSE index 5485559..87b582f 100644 --- a/LICENSE +++ b/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 diff --git a/contentstack/__init__.py b/contentstack/__init__.py index 233d0cc..8ce36ae 100644 --- a/contentstack/__init__.py +++ b/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__ = 'mobile@contentstack.com' __developer_email__ = 'shailesh.mishra@contentstack.com' diff --git a/contentstack/deep_merge_lp.py b/contentstack/deep_merge_lp.py index 1a0e58f..3970d65 100644 --- a/contentstack/deep_merge_lp.py +++ b/contentstack/deep_merge_lp.py @@ -1,17 +1,29 @@ 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, {}) @@ -19,3 +31,6 @@ def _deep_merge(self, source, destination): else: destination[key] = value return destination + + def to_dict(self): + return self.merged_response diff --git a/contentstack/entry.py b/contentstack/entry.py index a06d21a..260a52a 100644 --- a/contentstack/entry.py +++ b/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") + + + + diff --git a/contentstack/query.py b/contentstack/query.py index 8c4967e..4cbbc3e 100644 --- a/contentstack/query.py +++ b/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") \ No newline at end of file diff --git a/contentstack/stack.py b/contentstack/stack.py index 171fe17..21086f8 100644 --- a/contentstack/stack.py +++ b/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 diff --git a/requirements.txt b/requirements.txt index 0b7d2ab..bdefb12 100644 --- a/requirements.txt +++ b/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 diff --git a/tests/__init__.py b/tests/__init__.py index b7341fd..3545c68 100644 --- a/tests/__init__.py +++ b/tests/__init__.py @@ -12,6 +12,7 @@ from .test_entry import TestEntry from .test_query import TestQuery from .test_stack import TestStack +from .test_live_preview import TestLivePreviewConfig def all_tests(): @@ -19,9 +20,11 @@ def all_tests(): 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 ]) diff --git a/tests/test_live_preview.py b/tests/test_live_preview.py index deffb75..e9bc0b5 100644 --- a/tests/test_live_preview.py +++ b/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__':