- removed print() from tests directory

- changelog.rst latest version date updated

Author: ishaileshmishra

.github/workflows/publish.yml

@@ -0,0 +1,26 @@
+name: Publish to PyPI
+on:
+  release:
+    types: [ created ]
+jobs:
+  publish:
+    runs-on: ubuntu-latest
+    steps:
+      - name: Checkout
+        uses: actions/checkout@v2
+      - name: Set up Python
+        uses: actions/setup-python@v2
+        with:
+          python-version: '3.8'
+      - name: Install dependencies
+        run: |
+          python -m pip install --upgrade pip setuptools wheel
+          pip install -r requirements.txt
+      - name: Build package
+        run: python setup.py sdist bdist_wheel
+      - name: Publish package
+        uses: pypa/gh-action-pypi-publish@v1
+        with:
+          user: __token__
+          password: ${{ secrets.PYPI_PASSWORD }}
+          distributions: sdist bdist_wheel

CHANGELOG.md

@@ -2,11 +2,12 @@
 
 ## _v1.8.0_
 
-### **Date: 09-MAY-2023**
+### **Date: 23-MAY-2023**
 
-- AZURE_EU, region support added
-- include_metadata support added to asset, entry and query
+- AZURE_EU, Region support added
+- Include Metadata support added to asset, entry and query
 - General code improvement clean up
+- Updated code for Live Preview
 
 ---
 

changelog.rst

@@ -5,12 +5,12 @@
 *v1.8.0*
 ============
 
-**Date: 10-MAY-2023**
+**Date: 23-MAY-2023**
 
 - AZURE_EU, Region support added
 - Include Metadata support added to asset, entry and query
 - General code improvement clean up
-- Changes done for Live Preview
+- Updated code for Live Preview
 
 
 *v1.7.0*

contentstack/__init__.py

@@ -2,7 +2,6 @@
 The __init__.py files are required to make Python treat the directories as containing
 packages; this is done to prevent directories with a common name, such as string,
 from unintentionally hiding valid modules that occur later on the module search path
-
 """
 from .entry import Entry
 from .asset import Asset
@@ -14,7 +13,7 @@
 __title__ = 'contentstack-python'
 __author__ = 'contentstack'
 __status__ = 'debug'
-__version__ = '1.8.0'
+__version__ = 'v1.8.0'
 __endpoint__ = 'cdn.contentstack.io'
 __email__ = 'mobile@contentstack.com'
 __developer_email__ = 'shailesh.mishra@contentstack.com'

contentstack/deep_merge_lp.py

@@ -0,0 +1,21 @@
+class DeepMergeMixin:
+
+    def __init__(self, entry_response, lp_response):
+        self.entry_response = entry_response
+        self.lp_response = lp_response
+
+        for lp_obj in self.lp_response:
+            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)
+
+    def _deep_merge(self, source, destination):
+        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

contentstack/entry.py

@@ -8,6 +8,7 @@
 
 import empty
 
+from contentstack.deep_merge_lp import DeepMergeMixin
 from contentstack.entryqueryable import EntryQueryable
 
 log = logging.getLogger(__name__)
@@ -35,9 +36,9 @@ def __init__(self, http_instance, content_type_uid, entry_uid):
 
     def environment(self, environment):
         """
-        Enter the name of the environment of which the entries needs to be included
+        Enter the name of the environment of which the entries need to be included
         Example: production
-        :param environment: {str} name of the environment of which the entries needs to be included.
+        :param environment: {str} name of the environment of which the entries need to be included.
         :return: Entry, so you can chain this call.
         ------------------------------
         Example::
@@ -83,7 +84,7 @@ def param(self, key, value):
         This method is useful to add additional Query parameters to the entry
         :param key: {str} -- key The key as string which needs to be added to an Entry
         :param value: {object} -- value The value as string which needs to be added to an Entry
-        :return: Entry, so you can chain this call.
+        :return: @Entry, so you can chain this call.
         -----------------------------
         Example::
 
@@ -164,15 +165,6 @@ def __get_base_url(self, endpoint=''):
         url = f'{self.http_instance.endpoint}/content_types/{self.content_type_id}/entries/{self.entry_uid}'
         return url
 
-    def __validate_live_preview(self):
-        live_preview = self.http_instance.live_preview
-        if 'enable' in live_preview and live_preview['enable'] \
-                and self.content_type_id == live_preview['content_type_uid']:
-            if 'live_preview' in live_preview:
-                self.entry_param['live_preview'] = live_preview['live_preview']
-            else:
-                self.entry_param['live_preview'] = 'init'
-
     def fetch(self):
         """
         Fetches the latest version of the entries from stack
@@ -187,19 +179,35 @@ def fetch(self):
             >>> result = entry.fetch()
         -------------------------------
         """
-        self.__validate_live_preview()
         if 'environment' in self.http_instance.headers:
             self.entry_param['environment'] = self.http_instance.headers['environment']
         if len(self.entry_queryable_param) > 0:
             self.entry_param.update(self.entry_queryable_param)
-        encoded_string = parse.urlencode(self.entry_param, doseq=True)
-        url = f'{self.base_url}?{encoded_string}'
-        self._validate_live_preview()
-        return self.http_instance.get(url)
-
-    def _validate_live_preview(self):
-        lp = self.http_instance.live_preview
-        if 'content_type_uid' in lp and lp['content_type_uid'] is not None:
-            if lp['content_type_uid'] != str(self.content_type_id):
-                self.http_instance.live_preview['enable'] = False
+        encoded_str = parse.urlencode(self.entry_param, doseq=True)
+        url = f'{self.base_url}?{encoded_str}'
+        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['entry']
+            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_id:
+            url = lv['url']
+            self.http_instance.headers['authorization'] = lv['management_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
+            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']['entry']
+            lp_response = self.http_instance.live_preview['lp_response']
+            merged_response = DeepMergeMixin(entry_response, lp_response)
+            return merged_response.entry_response
         pass

contentstack/https_connection.py

@@ -2,9 +2,6 @@
 This module implements the Requests API.
 """
 
-# ************* Module https_connection.py **************
-# Your code has been rated at 10.00/10  by pylint
-
 import logging
 import platform
 import requests
@@ -28,13 +25,9 @@ def __get_os_platform():
 
 
 def user_agents():
-    """Default User Agents for the Https"""
-    header = {'sdk': dict(
-        name=contentstack.__package__,
-        version=contentstack.__version__
-    ),
-        'os': __get_os_platform,
-        'Content-Type': 'application/json'}
+    header = {'sdk': dict(name=contentstack.__package__,
+                          version=contentstack.__version__
+                          ), 'os': __get_os_platform, 'Content-Type': 'application/json'}
     package = f"{contentstack.__title__}/{contentstack.__version__}"
     return {'User-Agent': str(header), "X-User-Agent": package}
 
@@ -62,14 +55,22 @@ def __init__(self, endpoint, headers, timeout, retry_strategy, live_preview):
 
     def impl_live_preview(self):
         if self.live_preview['enable']:
-            print(self.live_preview)
-            # Get all the params from live preview and make a request,
-            # get the data and merger it to the base response
-        pass
+            host = self.live_preview['host']
+            authorization = self.live_preview['authorization']
+            ct = self.live_preview['content_type_uid']
+            entry_uid = self.live_preview['entry_uid']
+            url = f'https://{host}/v3/content_types/{ct}/entries'
+            if entry_uid is not None:
+                url = f'{url}/{entry_uid}'
+            self.headers['authorization'] = authorization
+            lp_resp = get_request(self.session, url, headers=self.headers, timeout=self.timeout)
+            if lp_resp is not None and not 'error_code' in lp_resp:
+                return lp_resp
+            return None
+        return None
 
     def get(self, url):
         self.headers.update(user_agents())
         adapter = HTTPAdapter(max_retries=self.retry_strategy)
         self.session.mount('https://', adapter)
-        self.impl_live_preview()
         return get_request(self.session, url, headers=self.headers, timeout=self.timeout)

contentstack/query.py

@@ -11,6 +11,7 @@
 import empty
 
 from contentstack.basequery import BaseQuery
+from contentstack.deep_merge_lp import DeepMergeMixin
 from contentstack.entryqueryable import EntryQueryable
 
 log = logging.getLogger(__name__)
@@ -262,9 +263,21 @@ def include_embedded_items(self):
         return self
 
     def include_metadata(self):
-        """
-        include_metadata includes metadata in the response
-        """
+        """include_metadata instance of Query
+                includes metadata in the response (Entries and Assets) along with entry/entries details.
+                :return: Query, so we can chain the call
+
+                ----------------------------
+                Example:
+
+                    >>> import contentstack
+                    >>> stack = contentstack.Stack('api_key', 'delivery_token', 'environment')
+                    >>> content_type = stack.content_type('content_type_uid')
+                    >>> query = content_type.query()
+                    >>> query = query.include_metadata()
+                    >>> result = query.find()
+                ----------------------------
+                """
         self.query_params['include_metadata'] = 'true'
         return self
 
@@ -303,15 +316,6 @@ def find_one(self):
         self.query_params["limit"] = 1
         return self.__execute_network_call()
 
-    def __validate_live_preview(self):
-        live_preview = self.http_instance.live_preview
-        if 'enable' in live_preview and live_preview['enable'] \
-                and self.content_type_uid == live_preview['content_type_uid']:
-            if 'live_preview' in live_preview:
-                self.query_params['live_preview'] = live_preview['live_preview']
-            else:
-                self.query_params['live_preview'] = 'init'  # initialise
-
     def __execute_network_call(self):
         if len(self.entry_queryable_param) > 0:
             self.query_params.update(self.entry_queryable_param)
@@ -321,29 +325,29 @@ def __execute_network_call(self):
             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}'
-        if self.http_instance.live_preview['enable']:
-            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)
-        self._validate_live_preview()
-        return self.http_instance.get(url)
-
-    def _validate_live_preview(self):
-        lp = self.http_instance.live_preview
-        if 'content_type_uid' in lp and lp['content_type_uid'] is not None:
-            if lp['content_type_uid'] != str(self.content_type_uid):
-                self.http_instance.live_preview['enable'] = False
+        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']
+            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:
+            url = lv['url']
+            self.http_instance.headers['authorization'] = lv['management_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
+            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']
+            merged_response = DeepMergeMixin(entry_response, lp_response)
+            return merged_response.entry_response
         pass
-
-    # 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 is properly written or not
-    #         else:
-    #             for key in dict.keys():
-    #                 qresp[key] = self._merge_preview(qresp[key])
-    #     elif isinstance(qresp, list):
-    #         for index, it in enumerate(qresp):
-    #             qresp[index] = self._merge_preview(it, _preview)
-    #     return qresp

contentstack/stack.py

@@ -341,7 +341,7 @@ def live_preview_query(self, **kwargs):
             for key, value in kwargs.iteritems():
                 print "%s = %s" % (key, value)
             Uses:=>
-        live_preview_query (
+        live_preview_query = (
                 'enable': True,
                 'live_preview': '#*#*#*#*#',
                 'host': 'your_host',
@@ -350,15 +350,28 @@ def live_preview_query(self, **kwargs):
                 'authorization': 'management_token'
                 )
         """
-        if self.live_preview is not None and self.live_preview['enable']:
-            query = kwargs['live_preview']
+
+        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()
         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