Repository: plone/plone.restapi (language: python)

Issue:
@aliases endpoint should allow to upload/download all aliases and allow to download filtered aliases as CSV file
PR for adding aliases: https://github.com/plone/plone.restapi/pull/1398

The @aliases should allow to upload aliases as CSV files and to download all existing aliases as CSV file. I am guessing there is a browser view in ClassicUI that does this already. Therefore it should be pretty straight forward to add this to the existing controlpanel. 

However, it would be nice if we would allow to download a CSV file based on any filter criterion of the @aliases endpoint:

http://localhost:8080/Plone/++api++/@aliases&q=/de/ft/my-path&manual=true&from=2021-01-01&to=2023-01-01

We could use content negotiation for this and send the "Accept" HTTP header "text/csv". Then REST API would return the CSV file.

# Code

Download method of the redirects control panel:

https://github.com/plone/Products.CMFPlone/blob/master/Products/CMFPlone/controlpanel/browser/redirects.py#L457

There is not a single unit/integration test for the aliases service. :(


Discussion / hints:

Fix patch:
diff --git a/docs/source/endpoints/aliases.md b/docs/source/endpoints/aliases.md
index 46f68342e1..da841c2e0a 100644
--- a/docs/source/endpoints/aliases.md
+++ b/docs/source/endpoints/aliases.md
@@ -70,9 +70,10 @@ Response:
 :language: http
 ```
 
-## Adding URL aliases in bulk
+## Adding URL aliases in bulk via JSON
 
-You can add multiple URL aliases for multiple pages by sending a `POST` request to the `/@aliases` endpoint on site `root`. **datetime** parameter is optional:
+You can add multiple URL aliases for multiple pages by sending a `POST` request to the `/@aliases` endpoint on site `root` using a JSON payload.
+**datetime** parameter is optional:
 
 ```{eval-rst}
 ..  http:example:: curl httpie python-requests
@@ -85,10 +86,26 @@ Response:
 :language: http
 ```
 
+## Adding URL aliases in bulk via CSV
 
-## Listing all available aliases
+You can add multiple URL aliases for multiple pages by sending a `POST` request to the `/@aliases` endpoint on site `root` using a CSV file.
+**datetime** parameter is optional:
 
-To list all aliases, send a `GET` request to the `/@aliases` endpoint on site `root`:
+```{eval-rst}
+..  http:example:: curl httpie python-requests
+    :request: ../../../src/plone/restapi/tests/http-examples/aliases_root_add_csv_format.req
+```
+
+Response:
+
+```{literalinclude} ../../../src/plone/restapi/tests/http-examples/aliases_root_add_csv_format.resp
+:language: http
+```
+
+
+## Listing all available aliases via JSON
+
+To list all aliases in JSON format, send a `GET` request to the `/@aliases` endpoint on site `root`:
 
 ```{eval-rst}
 ..  http:example:: curl httpie python-requests
@@ -101,6 +118,21 @@ Response:
 :language: http
 ```
 
+## Listing all available aliases via CSV
+
+To download all aliases as a CSV file, send a `GET` request to the `/@aliases` endpoint on site `root`:
+
+```{eval-rst}
+..  http:example:: curl httpie python-requests
+    :request: ../../../src/plone/restapi/tests/http-examples/aliases_root_get_csv_format.req
+```
+
+Response:
+
+```{literalinclude} ../../../src/plone/restapi/tests/http-examples/aliases_root_get_csv_format.resp
+:language: http
+```
+
 ## Filter aliases
 
 To search for specific aliases, send a `GET` request to the `/@aliases` endpoint on site `root` with a `q` parameter:
diff --git a/news/1812.feature b/news/1812.feature
new file mode 100644
index 0000000000..85727b674b
--- /dev/null
+++ b/news/1812.feature
@@ -0,0 +1,1 @@
+Added create and fetch aliases in CSV format. @Faakhir30
diff --git a/src/plone/restapi/services/aliases/add.py b/src/plone/restapi/services/aliases/add.py
index 555fbdbe0f..ab8af1ed2a 100644
--- a/src/plone/restapi/services/aliases/add.py
+++ b/src/plone/restapi/services/aliases/add.py
@@ -1,17 +1,24 @@
 from DateTime import DateTime
+from DateTime.interfaces import DateTimeError
 from plone.app.redirector.interfaces import IRedirectionStorage
 from plone.restapi import _
 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,15 +90,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")
             target = alias.get("redirect-to")
@@ -113,7 +140,11 @@ def reply(self):
 
             date = alias.get("datetime", None)
             if date:
-                date = DateTime(date)
+                try:
+                    date = DateTime(date)
+                except DateTimeError:
+                    logger.warning("Failed to parse as DateTime: %s", date)
+                    date = None
 
             storage.add(abs_redirection, abs_target, now=date, manual=True)
 
diff --git a/src/plone/restapi/services/aliases/configure.zcml b/src/plone/restapi/services/aliases/configure.zcml
index e3291cf161..c6499d117a 100644
--- a/src/plone/restapi/services/aliases/configure.zcml
+++ b/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"
diff --git a/src/plone/restapi/services/aliases/get.py b/src/plone/restapi/services/aliases/get.py
index c1fcf0b6ec..3ac3f81210 100644
--- a/src/plone/restapi/services/aliases/get.py
+++ b/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,24 +50,54 @@ 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()
+        filestream.close()
+
+        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"""
 
@@ -73,6 +105,16 @@ 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"""


Test patch:
diff --git a/src/plone/restapi/tests/http-examples/aliases_root_add_csv_format.req b/src/plone/restapi/tests/http-examples/aliases_root_add_csv_format.req
new file mode 100644
index 0000000000..fc049dcc0f
--- /dev/null
+++ b/src/plone/restapi/tests/http-examples/aliases_root_add_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=----WebKitFormBoundary7MA4YWxkTrZu0gW
+
+------WebKitFormBoundary7MA4YWxkTrZu0gW
+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
+
+------WebKitFormBoundary7MA4YWxkTrZu0gW--
diff --git a/src/plone/restapi/tests/http-examples/aliases_root_add_csv_format.resp b/src/plone/restapi/tests/http-examples/aliases_root_add_csv_format.resp
new file mode 100644
index 0000000000..0074ded3bc
--- /dev/null
+++ b/src/plone/restapi/tests/http-examples/aliases_root_add_csv_format.resp
@@ -0,0 +1,2 @@
+HTTP/1.1 204 No Content
+
diff --git a/src/plone/restapi/tests/http-examples/aliases_root_get_csv_format.req b/src/plone/restapi/tests/http-examples/aliases_root_get_csv_format.req
new file mode 100644
index 0000000000..be32ef6582
--- /dev/null
+++ b/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
diff --git a/src/plone/restapi/tests/http-examples/aliases_root_get_csv_format.resp b/src/plone/restapi/tests/http-examples/aliases_root_get_csv_format.resp
new file mode 100644
index 0000000000..04246f1761
--- /dev/null
+++ b/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
diff --git a/src/plone/restapi/tests/test_documentation.py b/src/plone/restapi/tests/test_documentation.py
index c25b896ef8..774bbd94b5 100644
--- a/src/plone/restapi/tests/test_documentation.py
+++ b/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,58 @@ 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_add_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"
+
+        # Setting a fixed boundary intentionally to make the producing .req and .resp files deterministic
+        boundary = "----WebKitFormBoundary7MA4YWxkTrZu0gW"
+
+        # Manually construct the multipart body
+        body = (
+            f"--{boundary}\r\n"
+            f'Content-Disposition: form-data; name="file"; filename="{csv_file.name}"\r\n'
+            "Content-Type: text/csv\r\n\r\n"
+            f"{content.decode()}\r\n"
+            f"--{boundary}--\r\n"
+        )
+
+        headers = {
+            "Accept": "application/json",
+            "Authorization": "Basic YWRtaW46c2VjcmV0",
+            "Content-Type": f"multipart/form-data; boundary={boundary}",
+        }
+
+        response = self.api_session.post(url, headers=headers, data=body)
+        save_request_and_response_for_docs("aliases_root_add_csv_format", response)
+
     def test_aliases_root_filter(self):
         # Get aliases
         url = f"{self.portal.absolute_url()}/@aliases"
diff --git a/src/plone/restapi/tests/test_services_aliases.py b/src/plone/restapi/tests/test_services_aliases.py
new file mode 100644
index 0000000000..27e4ca693e
--- /dev/null
+++ b/src/plone/restapi/tests/test_services_aliases.py
@@ -0,0 +1,162 @@
+from plone.app.testing import SITE_OWNER_PASSWORD
+from plone.app.testing import SITE_OWNER_NAME
+from plone.app.testing import TEST_USER_ID
+from plone.app.testing import setRoles
+from plone.restapi.testing import PLONE_RESTAPI_DX_FUNCTIONAL_TESTING
+from plone.restapi.testing import RelativeSession
+
+import transaction
+import unittest
+
+
+class TestAliases(unittest.TestCase):
+
+    layer = PLONE_RESTAPI_DX_FUNCTIONAL_TESTING
+
+    def setUp(self):
+        self.app = self.layer["app"]
+        self.portal = self.layer["portal"]
+        self.portal_url = self.portal.absolute_url()
+        setRoles(self.portal, TEST_USER_ID, ["Manager"])
+
+        self.api_session = RelativeSession(self.portal_url, test=self)
+        self.api_session.headers.update({"Accept": "application/json"})
+        self.api_session.auth = (SITE_OWNER_NAME, SITE_OWNER_PASSWORD)
+
+        self.portal.invokeFactory("Document", id="front-page")
+        transaction.commit()
+
+    def tearDown(self):
+        self.api_session.close()
+
+    def test_alias_non_root(self):
+        data = {
+            "items": [
+                {
+                    "path": "/alias-with-date",
+                    "redirect-to": "/front-page",
+                    "datetime": "2024-09-17T12:00:00",
+                }
+            ]
+        }
+        response = self.api_session.post("/front-page/@aliases", json=data)
+        self.assertEqual(response.status_code, 204)
+
+        # Verify alias exists
+        response = self.api_session.get("/front-page/@aliases")
+        self.assertEqual(response.status_code, 201)
+        self.assertEqual(len(response.json()["items"]), 1)
+
+    def test_alias_add_invalid_datetime(self):
+        """Test POST /@aliases with an invalid datetime, makes invalid date=None"""
+        data = {
+            "items": [
+                {
+                    "path": "/alias-with-valid-date",
+                    "redirect-to": "/front-page",
+                    "datetime": "2024-09-17T12:00:00",
+                },
+                {
+                    "path": "/alias-with-invalid-date",
+                    "redirect-to": "/front-page",
+                    "datetime": "invalid-date",
+                },
+            ]
+        }
+        response = self.api_session.post("/@aliases", json=data)
+        self.assertEqual(response.status_code, 204)
+        response = self.api_session.get("/@aliases")
+        self.assertEqual(response.status_code, 201)
+        self.assertEqual(len(response.json()["items"]), 2)
+
+    def test_alias_add_invalid_path(self):
+        """Test POST /@aliases with an invalid path"""
+
+        data = {"items": [{"path": "/valid-path", "redirect-to": "invalid-redirect"}]}
+        response = self.api_session.post("/@aliases", json=data)
+        self.assertEqual(response.status_code, 400)
+        response = self.api_session.get("/@aliases")
+        self.assertEqual(response.status_code, 201)
+        self.assertEqual(len(response.json()["items"]), 0)
+
+    def test_duplicate_alias(self):
+        data = {
+            "items": [
+                {"path": "/duplicate-alias", "redirect-to": "/front-page"},
+                {"path": "/duplicate-alias", "redirect-to": "/front-page"},
+            ]
+        }
+        self.api_session.post("/@aliases", json=data)
+        response = self.api_session.post("/@aliases", json=data)
+        self.assertEqual(response.status_code, 400)
+
+    def test_alias_without_redirect(self):
+        data = {"items": [{"path": "/alias-without-redirect"}]}
+        response = self.api_session.post("/@aliases", json=data)
+        self.assertEqual(response.status_code, 400)
+
+    def test_alias_csv_upload(self):
+        """Test POST /@aliases for CSV upload"""
+
+        content = b"old path,new path,datetime,manual\n/old-page,/front-page,2022/01/01 00:00:00 GMT+0,True\n"
+
+        response = self.api_session.post(
+            "/@aliases",
+            files={"file": ("aliases.csv", content, "text/csv")},
+        )
+
+        self.assertEqual(response.status_code, 204)
+        self.assertEqual(response.content, b"")
+        response = self.api_session.get("/@aliases")
+        self.assertEqual(response.status_code, 201)
+        self.assertEqual(
+            response.json().get("items"),
+            [
+                {
+                    "datetime": "2022-01-01T00:00:00+00:00",
+                    "manual": True,
+                    "path": "/old-page",
+                    "redirect-to": "/front-page",
+                }
+            ],
+        )
+
+    def test_alias_csv_download(self):
+        """Test GET /@aliases with CSV output"""
+
+        data = {
+            "items": [
+                {
+                    "path": "/alias-page",
+                    "redirect-to": "/front-page",
+                    "datetime": "2022/01/01 00:00:00 GMT+0",
+                },
+            ]
+        }
+        self.api_session.post("/@aliases", json=data)
+        headers = {"Accept": "text/csv"}
+        response = self.api_session.get("/@aliases", headers=headers)
+        self.assertEqual(response.status_code, 201)
+        self.assertIn("Content-Disposition", response.headers)
+        self.assertEqual(response.headers["Content-Type"], "text/csv; charset=utf-8")
+        content = b"old path,new path,datetime,manual\r\n/alias-page,/front-page,2022/01/01 00:00:00 GMT+0,True\r\n"
+        self.assertEqual(content, response.content)
+
+    def test_alias_delete(self):
+        data = {"items": [{"path": "/alias-to-delete", "redirect-to": "/front-page"}]}
+        self.api_session.post("/@aliases", json=data)
+        response = self.api_session.delete(
+            "/@aliases",
+            json={
+                "items": [
+                    {
+                        "path": "/alias-to-delete",
+                    }
+                ]
+            },
+        )
+        self.assertEqual(response.status_code, 204)
+
+        response = self.api_session.get("/@aliases")
+        self.assertEqual(response.status_code, 201)
+        self.assertEqual(len(response.json()["items"]), 0)