live_preview 🏆

Author: ishaileshmishra

contentstack/query.py

@@ -321,62 +321,20 @@ def __execute_network_call(self):
         self.__validate_live_preview()
         encoded_string = parse.urlencode(self.query_params, doseq=True)
         url = f'{self.base_url}?{encoded_string}'
-        # url = '{}?{}'.format(self.base_url, encoded_string)
-        if 'content_type_uid' in self.http_instance.live_preview and 'live_preview' in url:
-            query_response = self.http_instance.get(url)
-            return self._map_live_preview(query_response)
+        if self.http_instance.live_preview['content_type_uid'] == self.content_type_uid:
+            _rq = self.http_instance.get(url)['entries']
+            _preview = self.http_instance.live_preview['resp']
+            return self._merge_preview(_rq, _preview)
         return self.http_instance.get(url)
 
-    def _map_live_preview(self, query_response):
-        _preview = self.http_instance.live_preview['resp']  # Gets Live Preview Stored Data From Dict
-        if 'entries' in query_response:
-            query_response = query_response['entries']
-            for index, it in enumerate(query_response):  # it in query_response:
-                if isinstance(it, dict):
-                    if it['uid'] == _preview['uid']:
-                        # return self.merge_it(self.resp, _preview)
-                        merged = {**query_response[index], **_preview}
-                        return query_response
-                #   self._map_live_preview(query_response)
-                # elif isinstance(query_response, list):
-                #     for obj in query_response:
-                #         self.iter_dict(obj,_preview['uid'])
-                #     pass
-        else:
-            return None
-
-    # def iter_dict(self, item, uid):
-    #     if isinstance(item, dict):
-    #         if item['uid'] == uid:
-    #             # return self.merge_it(self.resp, preview_response)
-    #             merged = {**query_response[index], **preview_response}
-    #             return merged
-    #         self._map_live_preview(query_response)
-
-    def merge_it(self, resp, lp_resp):
-        resp.update(lp_resp)
-        # merged
-        merged = {**resp, **lp_resp}
-        print(f"merged: {merged}")
-        return resp
-
-    def deep_merge(self, resp, lp_resp):
-        query_set = set(resp)
-        live_set = set(lp_resp)
-        store_keys = []
-        for key in live_set.intersection(query_set):  # Iterates for common keys between query & livePreview
-            store_keys.append(key)
-            if isinstance(resp[key], dict):
-                # match keys and update value
-                self.merge_it(resp[key], lp_resp[key])
-            elif isinstance(resp[key], list):
-                # match keys and update value
-                for i in lp_resp[key]:
-                    if isinstance(resp[i], dict):
-                        self.merge_it(resp[key], lp_resp[key])
-                    resp.update({key: f"New Fake:{lp_resp[key]}"})
-            resp.update({key: f"New Fake:{lp_resp[key]}"})
-
-            # name, myNames[name]
-        print(f"new query resp: {resp}")
-        print(f"store_keys {store_keys}")
+    def _merge_preview(self, qresp, _preview):
+        if isinstance(qresp, dict):
+            if 'uid' in qresp and qresp['uid'] == _preview['uid']:
+                merged = {**qresp, **_preview}  # TODO: Check merging
+            else:
+                for key in dict.keys():
+                    qresp[key] = self._merge(qresp[key])
+        elif isinstance(qresp, list):
+            for index, it in enumerate(qresp):
+                qresp[index] = self._merge_preview(it, _preview)
+        return qresp

contentstack/stack.py

@@ -50,7 +50,7 @@ def __init__(self, api_key: str, delivery_token: str, environment: str,
                  branch=None,
                  ):
         """
-        Class that wraps the credentials of the authenticated user. Think of
+        # Class that wraps the credentials of the authenticated user. Think of
         this as a container that holds authentication related data.
 
         param api_key: api_key of the stack
@@ -65,13 +65,13 @@ def __init__(self, api_key: str, delivery_token: str, environment: str,
         takes input as dictionary object. containing one/multiple key value pair like below.
 
         ```python
-        live_preview = {
-            'enable': True,
+    live_preview = {
+              'enable': True,
             'authorization': 'your_management_token',
             'host': 'api.contentstack.com',
             'include_edit_tags': True,
             'edit_tags_type': object | str,
-        }
+    }
         ```
         :param retry_strategy: (optional) custom retry_strategy can be set.
         Method to create retry_strategy: create object of Retry() and provide the
@@ -101,20 +101,23 @@ def __init__(self, api_key: str, delivery_token: str, environment: str,
         self.timeout = timeout
         self.branch = branch
         self.retry_strategy = retry_strategy
-        self.live_preview_dict = live_preview
+        self.live_preview = live_preview
 
         self._validate_stack()
 
     def _validate_stack(self):
         if self.api_key is None or self.api_key == '':
             raise PermissionError(
-                'You are not permitted to the stack without valid APIKey')
+                'You are not permitted to the stack without valid APIKey'
+            )
         if self.delivery_token is None or self.delivery_token == "":
             raise PermissionError(
-                'You are not permitted to the stack without valid Delivery Token')
+                'You are not permitted to the stack without valid Delivery Token'
+            )
         if self.environment is None or self.environment == "":
             raise PermissionError(
-                'You are not permitted to the stack without valid Environment')
+                'You are not permitted to the stack without valid Environment'
+            )
 
         if self.region.value == 'eu' and self.host == DEFAULT_HOST:
             self.host = 'eu-cdn.contentstack.com'
@@ -133,24 +136,14 @@ def _validate_stack(self):
         if self.branch is not None:
             self.headers['branch'] = self.branch
 
-        if self.live_preview_dict is not None:
-            self._validate_live_preview()
         self.http_instance = HTTPSConnection(
             endpoint=self.endpoint,
-            headers=self.headers, timeout=self.timeout,
+            headers=self.headers,
+            timeout=self.timeout,
             retry_strategy=self.retry_strategy,
-            live_preview=self.live_preview_dict
+            live_preview=self.live_preview
         )
 
-    def _validate_live_preview(self):
-        if isinstance(self.live_preview_dict, dict):
-            if 'enable' in self.live_preview_dict and self.live_preview_dict['enable']:
-                self.headers['authorization'] = self.live_preview_dict['authorization']
-                self.host = self.live_preview_dict['host']
-                self.endpoint = f'https://{self.host}/{self.version}'
-                self.headers.pop('access_token')
-                self.headers.pop('environment')
-
     @property
     def get_api_key(self):
         """
@@ -191,7 +184,7 @@ def get_live_preview(self):
         """
         :return: live preview dictionary
         """
-        return self.live_preview_dict
+        return self.live_preview
 
     def content_type(self, content_type_uid=None):
         """
@@ -360,14 +353,28 @@ def live_preview_query(self, **kwargs):
                 hash='hashcode'
                 )
         """
-        self.live_preview_dict.update(kwargs)
+        self.live_preview.update(kwargs)
         self._execute_management_api()
         return self
 
     def _execute_management_api(self):
-        _ct_uid = self.live_preview_dict.get("content_type_uid")
-        _entry_uid = self.live_preview_dict.get("entry_uid")
-        _url = f'{self.http_instance.endpoint}/content_types/{_ct_uid}/entries/{_entry_uid}'
-        _resp = self.http_instance.get(_url)
-        self.live_preview_dict['resp'] = _resp['entry']
+        _headers, _endpoint = self._enable_live_preview()
+        _ct_uid = self.live_preview.get("content_type_uid")
+        _entry_uid = self.live_preview.get("entry_uid")
+        _url = f'{_endpoint}/content_types/{_ct_uid}/entries/{_entry_uid}'
+        import requests
+        r = requests.get(url=_url, verify=True, headers=_headers)
+        # _resp = self.http_instance.get(_url, headers=_headers)
+        self.live_preview['resp'] = r.json()['entry']
         return self
+
+    def _enable_live_preview(self):
+        if isinstance(self.live_preview, dict) and not None:
+            _endpoint = None
+            _local_headers = {}
+            if 'enable' in self.live_preview and self.live_preview['enable']:
+                _local_headers['authorization'] = self.live_preview['authorization']
+                _host = self.live_preview['host']
+                _endpoint = f'https://{_host}/{self.version}'
+                _local_headers['api_key'] = self.headers['api_key']
+            return _local_headers, _endpoint

tests/test_entry.py

@@ -156,5 +156,6 @@ def test_21_entry_include_embedded_items(self):
         self.assertEqual({'environment': 'development',
                           'include_embedded_items[]': 'BASE'}, entry.entry_param)
 
-# if __name__ == '__main__':
-#     unittest.main()
+
+if __name__ == '__main__':
+    unittest.main()

tests/test_live_preview_config.py tests/test_live_preview.pytests/test_live_preview_config.py renamed to tests/test_live_preview.py

@@ -41,26 +41,26 @@ def test_021_live_preview_enabled_(self):
             ENVIRONMENT,
             live_preview=_preview)
         self.assertEqual(_preview['authorization'],
-                         self.stack.live_preview_dict['authorization'])
+                         self.stack.live_preview['authorization'])
 
     def test_03_set_host(self):
         self.stack = contentstack.Stack(
             API_KEY,
             DELIVERY_TOKEN,
             ENVIRONMENT,
             live_preview=_preview)
-        self.assertEqual(3, len(self.stack.live_preview_dict))
-        self.assertEqual(True, self.stack.live_preview_dict['enable'])
+        self.assertEqual(3, len(self.stack.live_preview))
+        self.assertEqual(True, self.stack.live_preview['enable'])
 
     def test_031_set_host_value(self):
         self.stack = contentstack.Stack(
             API_KEY,
             DELIVERY_TOKEN,
             ENVIRONMENT,
             live_preview=_preview)
-        self.assertEqual(3, len(self.stack.live_preview_dict))
+        self.assertEqual(3, len(self.stack.live_preview))
         self.assertEqual(_preview['host'],
-                         self.stack.live_preview_dict['host'])
+                         self.stack.live_preview['host'])
 
     def test_06_live_preview_query(self):
         _live_preview = {
@@ -74,7 +74,7 @@ def test_06_live_preview_query(self):
             ENVIRONMENT,
             live_preview=_preview
         )
-        self.assertEqual(5, len(self.stack.live_preview_dict))
+        self.assertEqual(5, len(self.stack.live_preview))
 
     def test_07_live_preview_query_hash_included(self):
         self.stack = contentstack.Stack(
@@ -86,7 +86,7 @@ def test_07_live_preview_query_hash_included(self):
         self.stack.live_preview_query(
             hash='live_preview',
             content_type_uid='fake@content_type')
-        self.assertEqual(7, len(self.stack.live_preview_dict))
+        self.assertEqual(7, len(self.stack.live_preview))
 
     def test_08_live_preview_query_hash_excluded(self):
         self.stack = contentstack.Stack(
@@ -125,6 +125,24 @@ def test_branching(self):
         stack.content_type('product')
         self.assertEqual('dev_branch', stack.get_branch)
 
+    def test_live_preview(self):
+        __preview = {
+            'enable': True,
+            'authorization': 'management_token',
+            'host': 'api.contentstack.io'
+        }
+        stack = contentstack.Stack(
+            'api_key',
+            'delivery_token',
+            'development',
+            live_preview=__preview).live_preview_query(
+            entry_uid='entry_uid',
+            content_type_uid='bugfixes',
+            hash='#Just_fake_it'
+        )
+        result = stack.content_type(content_type_uid='bugfixes').query().find()
+        print(result)
+
 
 if __name__ == '__main__':
     unittest.main()