added TrashBin API (#106)

- TrashBin API:
  * `trashbin_list`
  * `trashbin_restore`
  * `trashbin_delete`
  * `trashbin_cleanup`

---------

Signed-off-by: Alexander Piskun <bigcat88@icloud.com>

Author: bigcat88

CHANGELOG.md

@@ -2,7 +2,15 @@
 
 All notable changes to this project will be documented in this file.
 
-## [0.0.42 - 2023-08-30]
+## [0.0.42 - 2023-08-3x]
+
+### Added
+
+- TrashBin API:
+  * `trashbin_list`
+  * `trashbin_restore`
+  * `trashbin_delete`
+  * `trashbin_cleanup`
 
 ### Fixed
 

README.md

@@ -36,7 +36,7 @@ Python library that provides a robust and well-documented API that allows develo
 | Text Processing** |     N/A      |      ❌       |      ❌       |
 | SpeechToText**    |     N/A      |      ❌       |      ❌       |
 
-&ast;missing `Trash bin` and `File version` support.<br>
+&ast;missing `File version` support.<br>
 &ast;&ast;available only for NextcloudApp
 
 ### Differences between the Nextcloud and NextcloudApp classes

nc_py_api/_version.py

@@ -1,3 +1,3 @@
 """Version of nc_py_api."""
 
-__version__ = "0.0.41"
+__version__ = "0.0.42.dev0"

nc_py_api/files/__init__.py

@@ -22,6 +22,7 @@ class FsNodeInfo:
     fileid: int
     """Clear file ID without Nextcloud instance ID."""
     _last_modified: datetime.datetime
+    _trashbin: dict
 
     def __init__(self, **kwargs):
         self.size = kwargs.get("size", 0)
@@ -33,6 +34,10 @@ def __init__(self, **kwargs):
             self.last_modified = kwargs.get("last_modified", datetime.datetime(1970, 1, 1))
         except (ValueError, TypeError):
             self.last_modified = datetime.datetime(1970, 1, 1)
+        self._trashbin: dict[str, typing.Union[str, int]] = {}
+        for i in ("trashbin_filename", "trashbin_original_location", "trashbin_deletion_time"):
+            if i in kwargs:
+                self._trashbin[i] = kwargs[i]
 
     @property
     def last_modified(self) -> datetime.datetime:
@@ -49,6 +54,26 @@ def last_modified(self, value: typing.Union[str, datetime.datetime]):
         else:
             self._last_modified = value
 
+    @property
+    def in_trash(self) -> bool:
+        """Returns ``True`` if the object is in trash."""
+        return bool(self._trashbin)
+
+    @property
+    def trashbin_filename(self) -> str:
+        """Returns the name of the object in the trashbin."""
+        return self._trashbin.get("trashbin_filename", "")
+
+    @property
+    def trashbin_original_location(self) -> str:
+        """Returns the original path of the object."""
+        return self._trashbin.get("trashbin_original_location", "")
+
+    @property
+    def trashbin_deletion_time(self) -> int:
+        """Returns deletion time of the object."""
+        return int(self._trashbin.get("trashbin_deletion_time", 0))
+
 
 @dataclasses.dataclass
 class FsNode:

nc_py_api/files/files.py

@@ -326,17 +326,65 @@ def setfav(self, path: Union[str, FsNode], value: Union[int, bool]) -> None:
         )
         check_error(webdav_response.status_code, f"setfav: path={path}, value={value}")
 
-    def _listdir(self, user: str, path: str, properties: list[str], depth: int, exclude_self: bool) -> list[FsNode]:
+    def trashbin_list(self) -> list[FsNode]:
+        """Returns a list of all entries in the TrashBin."""
+        properties = PROPFIND_PROPERTIES
+        properties += ["nc:trashbin-filename", "nc:trashbin-original-location", "nc:trashbin-deletion-time"]
+        return self._listdir(self._session.user, "", properties=properties, depth=1, exclude_self=False, trashbin=True)
+
+    def trashbin_restore(self, path: Union[str, FsNode]) -> None:
+        """Restore a file/directory from the TrashBin.
+
+        :param path: path to delete, e.g., the ``user_path`` field from ``FsNode`` or the **FsNode** class itself.
+        """
+        restore_name = path.name if isinstance(path, FsNode) else path.split("/", maxsplit=1)[-1]
+        path = path.user_path if isinstance(path, FsNode) else path
+
+        dest = self._session.cfg.dav_endpoint + f"/trashbin/{self._session.user}/restore/{restore_name}"
+        headers = {"Destination": dest}
+        response = self._session.dav(
+            "MOVE",
+            path=f"/trashbin/{self._session.user}/{path}",
+            headers=headers,
+        )
+        check_error(response.status_code, f"trashbin_restore: user={self._session.user}, src={path}, dest={dest}")
+
+    def trashbin_delete(self, path: Union[str, FsNode], not_fail=False) -> None:
+        """Deletes a file/directory permanently from the TrashBin.
+
+        :param path: path to delete, e.g., the ``user_path`` field from ``FsNode`` or the **FsNode** class itself.
+        :param not_fail: if set to ``True`` and the object is not found, it does not raise an exception.
+        """
+        path = path.user_path if isinstance(path, FsNode) else path
+        response = self._session.dav(method="DELETE", path=f"/trashbin/{self._session.user}/{path}")
+        if response.status_code == 404 and not_fail:
+            return
+        check_error(response.status_code, f"delete_from_trashbin: user={self._session.user}, path={path}")
+
+    def trashbin_cleanup(self) -> None:
+        """Empties the TrashBin."""
+        response = self._session.dav(method="DELETE", path=f"/trashbin/{self._session.user}/trash")
+        check_error(response.status_code, f"trashbin_cleanup: user={self._session.user}")
+
+    def _listdir(
+        self, user: str, path: str, properties: list[str], depth: int, exclude_self: bool, trashbin: bool = False
+    ) -> list[FsNode]:
         root = ElementTree.Element(
             "d:propfind",
             attrib={"xmlns:d": "DAV:", "xmlns:oc": "http://owncloud.org/ns", "xmlns:nc": "http://nextcloud.org/ns"},
         )
         prop = ElementTree.SubElement(root, "d:prop")
         for i in properties:
             ElementTree.SubElement(prop, i)
-        headers = {"Depth": "infinity" if depth == -1 else str(depth)}
+        if trashbin:
+            dav_path = self._dav_get_obj_path(f"trashbin/{user}/trash", path, root_path="")
+        else:
+            dav_path = self._dav_get_obj_path(user, path)
         webdav_response = self._session.dav(
-            "PROPFIND", self._dav_get_obj_path(user, path), data=self._element_tree_as_str(root), headers=headers
+            "PROPFIND",
+            dav_path,
+            self._element_tree_as_str(root),
+            headers={"Depth": "infinity" if depth == -1 else str(depth)},
         )
         request_info = f"list: {user}, {path}, {properties}"
         result = self._lf_parse_webdav_records(webdav_response, request_info)
@@ -387,6 +435,12 @@ def _parse_record(full_path: str, prop_stats: list[dict]) -> FsNode:
                 fs_node_args["permissions"] = prop["oc:permissions"]
             if "oc:favorite" in prop_keys:
                 fs_node_args["favorite"] = bool(int(prop["oc:favorite"]))
+            if "nc:trashbin-filename" in prop_keys:
+                fs_node_args["trashbin_filename"] = prop["nc:trashbin-filename"]
+            if "nc:trashbin-original-location" in prop_keys:
+                fs_node_args["trashbin_original_location"] = prop["nc:trashbin-original-location"]
+            if "nc:trashbin-deletion-time" in prop_keys:
+                fs_node_args["trashbin_deletion_time"] = prop["nc:trashbin-deletion-time"]
             # xz = prop.get("oc:dDC", "")
         return FsNode(full_path, **fs_node_args)
 

tests/files_test.py

@@ -594,3 +594,49 @@ def test_fs_node_last_modified_time():
     assert fs_node.info.last_modified == datetime(2023, 7, 29, 11, 56, 31)
     fs_node = FsNode("", last_modified=datetime(2022, 4, 5, 1, 2, 3))
     assert fs_node.info.last_modified == datetime(2022, 4, 5, 1, 2, 3)
+
+
+def test_trashbin(nc):
+    r = nc.files.trashbin_list()
+    assert isinstance(r, list)
+    new_file = nc.files.upload("nc_py_api_temp.txt", content=b"")
+    nc.files.delete(new_file)
+    # minimum one object now in a trashbin
+    r = nc.files.trashbin_list()
+    assert r
+    # clean up trashbin
+    nc.files.trashbin_cleanup()
+    # no objects should be in trashbin
+    r = nc.files.trashbin_list()
+    assert not r
+    new_file = nc.files.upload("nc_py_api_temp.txt", content=b"")
+    nc.files.delete(new_file)
+    # one object now in a trashbin
+    r = nc.files.trashbin_list()
+    assert len(r) == 1
+    # check properties types of FsNode
+    i: FsNode = r[0]
+    assert i.info.in_trash is True
+    assert i.info.trashbin_filename.find("nc_py_api_temp.txt") != -1
+    assert i.info.trashbin_original_location == "nc_py_api_temp.txt"
+    assert isinstance(i.info.trashbin_deletion_time, int)
+    # restore that object
+    nc.files.trashbin_restore(r[0])
+    # no files in trashbin
+    r = nc.files.trashbin_list()
+    assert not r
+    # move a restored object to trashbin again
+    nc.files.delete(new_file)
+    # one object now in a trashbin
+    r = nc.files.trashbin_list()
+    assert len(r) == 1
+    # remove one object from a trashbin
+    nc.files.trashbin_delete(r[0])
+    # NextcloudException with status_code 404
+    with pytest.raises(NextcloudException) as e:
+        nc.files.trashbin_delete(r[0])
+    assert e.value.status_code == 404
+    nc.files.trashbin_delete(r[0], not_fail=True)
+    # no files in trashbin
+    r = nc.files.trashbin_list()
+    assert not r