Merge pull request #381 from palewire/create-folder

datawrapper/__main__.py

@@ -41,7 +41,7 @@ class Datawrapper:
     _BASE_URL = "https://api.datawrapper.de"
     _CHARTS_URL = _BASE_URL + "/v3/charts"
     _PUBLISH_URL = _BASE_URL + "/charts"
-    _FOLDERS_URL = _BASE_URL + "/folders"
+    _FOLDERS_URL = _BASE_URL + "/v3/folders"
 
     _ACCESS_TOKEN = os.getenv("DATAWRAPPER_ACCESS_TOKEN")
 
@@ -620,6 +620,169 @@ def get_folders(self) -> dict[Any, Any] | None | Any:
             logger.error(msg)
             raise Exception(msg)
 
+    def get_folder(self, folder_id: str | int) -> dict[Any, Any]:
+        """Get an existing folder.
+
+        Parameters
+        ----------
+        folder_id : str | int
+            ID of folder to get.
+
+        Returns
+        -------
+        dict
+            A dictionary containing the folder's information.
+        """
+        _header = self._auth_header
+        _header["accept"] = "*/*"
+
+        response = r.get(
+            url=self._FOLDERS_URL + f"/{folder_id}",
+            headers=_header,
+        )
+
+        if response.ok:
+            folder_info = response.json()
+            logger.debug(f"Folder {folder_info['name']} retrieved with id {folder_id}")
+            return folder_info
+        else:
+            msg = "Folder could not be retrieved."
+            logger.error(msg)
+            raise Exception(msg)
+
+    def create_folder(
+        self,
+        name: str,
+        parent_id: str | int | None = None,
+        team_id: str | int | None = None
+    ) -> dict[Any, Any]:
+        """Create a new folder.
+
+        Parameters
+        ----------
+        name: str
+            Name of the folder to be created.
+        parent_id: str | int, optional
+            The parent folder that the folder belongs to.
+        team_id: str | int, optional
+            The team that the folder belongs to. If teamId is empty, the folder will belong to the user directly.
+
+        Returns
+        -------
+        dict
+            A dictionary containing the folder's information.
+        """
+        _header = self._auth_header
+        _header["accept"] = "*/*"
+
+        _query: dict[str, Any] = {"name": name}
+        if parent_id:
+            _query["parentId"] = parent_id
+        if team_id:
+            _query["teamId"] = team_id
+
+        response = r.post(
+            url=self._FOLDERS_URL,
+            headers=_header,
+            data=json.dumps(_query),
+        )
+
+        if response.ok:
+            folder_info = response.json()
+            logger.debug(f"Folder {folder_info['name']} created with id {folder_info['id']}")
+            return folder_info
+        else:
+            msg = "Folder could not be created."
+            logger.error(msg)
+            raise Exception(msg)
+
+    def update_folder(
+        self,
+        folder_id: str | int,
+        name: str | None = None,
+        parent_id: str | int | None = None,
+        team_id: str | int | None = None,
+        user_id: str | int | None = None,
+    ) -> dict[Any, Any]:
+        """Update an existing folder.
+
+        Parameters
+        ----------
+        folder_id : str | int
+            ID of folder to update.
+        name: str, optional
+            Name to change the folder to.
+        parent_id: str | int, optional
+            The parent folder where this folder is stored.
+        team_id: str | int, optional
+            The team that the folder belongs to.
+        user_id: str | int, optional
+            The user that the folder belongs to.
+
+        Returns
+        -------
+        r.Response.content
+            The content of the requests.delete
+        """
+        _header = self._auth_header
+        _header["accept"] = "*/*"
+
+        _query: dict[str, Any] = {}
+        if name:
+            _query["name"] = name
+        if parent_id:
+            _query["parentId"] = parent_id
+        if team_id:
+            _query["teamId"] = team_id
+        if user_id:
+            _query["userId"] = user_id
+
+        url = self._FOLDERS_URL + f"/{folder_id}"
+        response = r.patch(
+            url=url,
+            headers=_header,
+            data=json.dumps(_query),
+        )
+
+        if response.ok:
+            folder_info = response.json()
+            logger.debug(f"Folder {folder_id} updated")
+            return folder_info
+        else:
+            msg = "Folder could not be updated."
+            logger.error(msg)
+            raise Exception(msg)
+
+    def delete_folder(self, folder_id: str | int):
+        """Delete an existing folder.
+
+        Parameters
+        ----------
+        folder_id : str | int
+            ID of folder to delete.
+
+        Returns
+        -------
+        r.Response.content
+            The content of the requests.delete
+        """
+        _header = self._auth_header
+        _header["accept"] = "*/*"
+
+        url = self._FOLDERS_URL + f"/{folder_id}"
+        response = r.delete(
+            url=url,
+            headers=_header,
+        )
+
+        if response.ok:
+            logger.debug(f"Folder {folder_id} deleted")
+            return response.content
+        else:
+            msg = "Folder could not be deleted."
+            logger.error(msg)
+            raise Exception(msg)
+
     def move_chart(self, chart_id: str, folder_id: str) -> Any | None:
         """Moves a chart, table, or map to a specified folder.
 

tests/test_folders.py

@@ -0,0 +1,58 @@
+"""Test folder related API enpoints."""
+import random
+import string
+
+import pytest
+
+from datawrapper import Datawrapper
+
+
+def test_get_folders():
+    """Test the get_folders method."""
+    dw = Datawrapper()
+    folder_list = dw.get_folders()
+    assert len(folder_list) > 0
+
+
+def test_folder_crud():
+    """Run folder related tests for creation, updating and deleting."""
+    # Connect
+    dw = Datawrapper()
+
+    # Get a randoms string suffix to use in our names
+    suffix = "".join(random.choices(string.ascii_lowercase + string.digits, k=5))
+
+    # Create a new folder
+    folder_info = dw.create_folder(name="My new folder " + suffix)
+
+    # Get the folder's data with a fresh get_folder call
+    folder_info = dw.get_folder(folder_info["id"])
+
+    # Make a second folder
+    second_folder_info = dw.create_folder(name="My second folder " + suffix)
+
+    # Move the second folder into the first folder
+    dw.update_folder(
+        folder_id=second_folder_info["id"], parent_id=folder_info["id"]
+    )
+
+    # Get the folder's data with a fresh get_folder call
+    second_folder_info = dw.get_folder(second_folder_info["id"])
+
+    # Verify it has the parent
+    assert second_folder_info["parentId"] == folder_info["id"]
+
+    # Change the name of the second folder
+    dw.update_folder(folder_id=second_folder_info["id"], name="My second folder (renamed)")
+
+    # Get it fresh and verify the change
+    second_folder_info = dw.get_folder(second_folder_info["id"])
+    assert second_folder_info["name"] == "My second folder (renamed)"
+
+    # Delete both folders
+    dw.delete_folder(folder_info["id"])
+
+    # Verify that you can't get either
+    with pytest.raises(Exception):
+        dw.get_folder(folder_info["id"])
+        dw.get_folder(second_folder_info["id"])