Added create and fetch aliases using csv.

news/1812.feature

@@ -0,0 +1 @@
+Added create and fetch aliases in CSV format. @Faakhir30

src/plone/restapi/services/aliases/add.py

@@ -4,14 +4,20 @@
 from plone.restapi.deserializer import json_body
 from plone.restapi.services import Service
 from Products.CMFPlone.controlpanel.browser.redirects import absolutize_path
+from Products.CMFPlone.controlpanel.browser.redirects import RedirectsControlPanel
+from Products.statusmessages.interfaces import IStatusMessage
 from zExceptions import BadRequest
 from zope.component import getMultiAdapter
+from zope.component.hooks import getSite
 from zope.component import getUtility
 from zope.interface import alsoProvides
 from zope.interface import implementer
 from zope.publisher.interfaces import IPublishTraverse
 
 import plone.protect.interfaces
+import logging
+
+logger = logging.getLogger("Plone")
 
 
 @implementer(IPublishTraverse)
@@ -83,14 +89,35 @@ def edit_for_navigation_root(self, alias):
 class AliasesRootPost(Service):
     """Creates new aliases via controlpanel"""
 
-    def reply(self):
-        data = json_body(self.request)
+    def _reply_csv(self):
+        form = self.request.form
+        if not form.get("file"):
+            raise BadRequest("No file uploaded")
+        controlpanel = RedirectsControlPanel(self.context, self.request)
         storage = getUtility(IRedirectionStorage)
-        aliases = data.get("items", [])
+        status = IStatusMessage(self.request)
+        portal = getSite()
+        file = form["file"]
+        controlpanel.upload(file, portal, storage, status)
+        file.close()
+
+        if err := status.show():
+            if err[0].type == "error":
+                raise BadRequest(err[0].message)
+            elif err[0].type == "info":
+                logger.info(err[0].message)
+        return self.reply_no_content()
 
+    def reply(self):
         # Disable CSRF protection
         if "IDisableCSRFProtection" in dir(plone.protect.interfaces):
             alsoProvides(self.request, plone.protect.interfaces.IDisableCSRFProtection)
+        if "multipart/form-data" in self.request.getHeader("Content-Type"):
+            return self._reply_csv()
+
+        storage = getUtility(IRedirectionStorage)
+        data = json_body(self.request)
+        aliases = data.get("items", [])
 
         for alias in aliases:
             redirection = alias.get("path")

src/plone/restapi/services/aliases/configure.zcml

@@ -12,6 +12,15 @@
       name="@aliases"
       />
 
+  <plone:service
+      method="GET"
+      accept="text/csv"
+      factory=".get.AliasesGet"
+      for="Products.CMFPlone.interfaces.IPloneSiteRoot"
+      permission="zope2.View"
+      name="@aliases"
+      />
+
   <plone:service
       method="GET"
       accept="application/json,application/schema+json"

src/plone/restapi/services/aliases/get.py

@@ -9,6 +9,7 @@
 from zope.component.hooks import getSite
 from zope.interface import implementer
 from zope.interface import Interface
+import json
 
 
 @implementer(IExpandableElement)
@@ -26,7 +27,8 @@ def reply_item(self):
         redirects = storage.redirects(context_path)
         aliases = [deroot_path(alias) for alias in redirects]
         self.request.response.setStatus(201)
-        return [{"path": alias} for alias in aliases]
+        self.request.response.setHeader("Content-Type", "application/json")
+        return [{"path": alias} for alias in aliases], len(aliases)
 
     def reply_root(self):
         """
@@ -48,31 +50,69 @@ def reply_root(self):
 
         newbatch = RedirectsControlPanel(self.context, self.request).redirects()
         items_total = len([item for item in newbatch])
+        self.request.response.setHeader("Content-Type", "application/json")
+
         return redirects, items_total
 
+    def reply_root_csv(self):
+        batch = RedirectsControlPanel(self.context, self.request).redirects()
+        redirects = [entry for entry in batch]
+
+        for redirect in redirects:
+            del redirect["redirect"]
+            redirect["datetime"] = datetimelike_to_iso(redirect["datetime"])
+        self.request.response.setStatus(201)
+
+        self.request.form["b_start"] = "0"
+        self.request.form["b_size"] = "1000000"
+        self.request.__annotations__.pop("plone.memoize")
+
+        filestream = RedirectsControlPanel(self.context, self.request).download()
+        content = filestream.read()
+        self.request.response.setHeader("Content-Type", "text/csv")
+        self.request.response.setHeader(
+            "Content-Disposition", "attachment; filename=redirects.csv"
+        )
+        self.request.response.setHeader("Content-Length", str(len(content)))
+        return content
+
     def __call__(self, expand=False):
         result = {"aliases": {"@id": f"{self.context.absolute_url()}/@aliases"}}
         if not expand:
             return result
-
         if IPloneSiteRoot.providedBy(self.context):
-            items, items_total = self.reply_root()
-            result["aliases"]["items"] = items
-            result["aliases"]["items_total"] = items_total
+            if self.request.getHeader("Accept") == "text/csv":
+                result["aliases"]["items"] = self.reply_root_csv()
+                return result
+            else:
+                items, items_total = self.reply_root()
         else:
-            result["aliases"]["items"] = self.reply_item()
-            result["aliases"]["items_total"] = len(result["aliases"]["items"])
-
+            items, items_total = self.reply_item()
+        result["aliases"]["items"] = items
+        result["aliases"]["items_total"] = items_total
         return result
 
 
+_no_content_marker = object()
+
+
 class AliasesGet(Service):
     """Get aliases"""
 
     def reply(self):
         aliases = Aliases(self.context, self.request)
         return aliases(expand=True)["aliases"]
 
+    def render(self):
+        self.check_permission()
+        content = self.reply()
+        if self.request.getHeader("Accept") == "text/csv":
+            return content["items"]
+        if content is not _no_content_marker:
+            return json.dumps(
+                content, indent=2, sort_keys=True, separators=(", ", ": ")
+            )
+
 
 def deroot_path(path):
     """Remove the portal root from alias"""

src/plone/restapi/tests/http-examples/aliases_root_get_csv_format.req

@@ -0,0 +1,3 @@
+GET /plone/@aliases HTTP/1.1
+Accept: text/csv
+Authorization: Basic YWRtaW46c2VjcmV0

src/plone/restapi/tests/http-examples/aliases_root_get_csv_format.resp

@@ -0,0 +1,6 @@
+HTTP/1.1 201 Created
+Content-Type: text/csv; charset=utf-8
+
+old path,new path,datetime,manual
+/fizzbuzz,/front-page,2022/05/05 00:00:00 GMT+0,True
+/old-page,/front-page,2022/05/05 00:00:00 GMT+0,True

src/plone/restapi/tests/http-examples/aliases_root_post_csv_format.req

@@ -0,0 +1,13 @@
+POST /plone/@aliases HTTP/1.1
+Accept: application/json
+Authorization: Basic YWRtaW46c2VjcmV0
+Content-Type: multipart/form-data; boundary=8ef8f6e729077f0afdcee18b29387cf7
+
+--8ef8f6e729077f0afdcee18b29387cf7
+Content-Disposition: form-data; name="file"; filename="test_file.csv"
+Content-Type: text/csv
+
+old path,new path,datetime,manual
+/old-page,/front-page,2022/01/01 00:00:00 GMT+0,True
+
+--8ef8f6e729077f0afdcee18b29387cf7--

src/plone/restapi/tests/http-examples/aliases_root_post_csv_format.resp

@@ -0,0 +1,2 @@
+HTTP/1.1 204 No Content
+

src/plone/restapi/tests/test_documentation.py

@@ -1,6 +1,7 @@
 from base64 import b64encode
 from datetime import datetime
 from datetime import timezone
+import io
 from pkg_resources import resource_filename
 from plone import api
 from plone.app.discussion.interfaces import ICommentAddedEvent
@@ -2097,6 +2098,42 @@ def test_aliases_root_get(self):
         response = self.api_session.get(url + query)
         save_request_and_response_for_docs("aliases_root_get", response)
 
+    def test_aliases_root_get_csv_format(self):
+        url = f"{self.portal.absolute_url()}/@aliases"
+        query = ""
+
+        payload = {
+            "items": [
+                {
+                    "path": "/old-page",
+                    "redirect-to": "/front-page",
+                    "datetime": "2022-05-05",
+                },
+                {
+                    "path": "/fizzbuzz",
+                    "redirect-to": "/front-page",
+                    "datetime": "2022-05-05",
+                },
+            ]
+        }
+        response = self.api_session.post(url, json=payload)
+        self.api_session.headers.update({"Content-Type": "application/json"})
+        self.api_session.headers.update({"Accept": "text/csv"})
+        response = self.api_session.get(url + query)
+        save_request_and_response_for_docs("aliases_root_get_csv_format", response)
+
+    def test_aliases_root_post_csv_format(self):
+        url = f"{self.portal.absolute_url()}/@aliases"
+
+        content = b"old path,new path,datetime,manual\n/old-page,/front-page,2022/01/01 00:00:00 GMT+0,True\n"
+        csv_file = io.BytesIO(content)
+        csv_file.name = "test_file.csv"
+
+        files = {"file": (csv_file.name, csv_file, "text/csv")}
+        self.api_session.headers.update({"Accept": "application/json"})
+        response = self.api_session.post(url, files=files)
+        save_request_and_response_for_docs("aliases_root_post_csv_format", response)
+
     def test_aliases_root_filter(self):
         # Get aliases
         url = f"{self.portal.absolute_url()}/@aliases"