From ef33ab4d453927cebf437c77779c409c2fb21d4f Mon Sep 17 00:00:00 2001 From: Alexander Piskun <13381981+bigcat88@users.noreply.github.com> Date: Tue, 29 Aug 2023 19:45:32 +0300 Subject: [PATCH] added TrashBin API (#106) - TrashBin API: * `trashbin_list` * `trashbin_restore` * `trashbin_delete` * `trashbin_cleanup` --------- Signed-off-by: Alexander Piskun --- CHANGELOG.md | 10 ++++++- README.md | 2 +- nc_py_api/_version.py | 2 +- nc_py_api/files/__init__.py | 25 ++++++++++++++++ nc_py_api/files/files.py | 60 +++++++++++++++++++++++++++++++++++-- tests/files_test.py | 46 ++++++++++++++++++++++++++++ 6 files changed, 139 insertions(+), 6 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 9d7026fc..6b727a10 100644 --- a/CHANGELOG.md +++ b/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 diff --git a/README.md b/README.md index 1688d7cc..60c944de 100644 --- a/README.md +++ b/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 | ❌ | ❌ | -*missing `Trash bin` and `File version` support.
+*missing `File version` support.
**available only for NextcloudApp ### Differences between the Nextcloud and NextcloudApp classes diff --git a/nc_py_api/_version.py b/nc_py_api/_version.py index cb980a6f..651a3e8d 100644 --- a/nc_py_api/_version.py +++ b/nc_py_api/_version.py @@ -1,3 +1,3 @@ """Version of nc_py_api.""" -__version__ = "0.0.41" +__version__ = "0.0.42.dev0" diff --git a/nc_py_api/files/__init__.py b/nc_py_api/files/__init__.py index 4eab78f3..773955f0 100644 --- a/nc_py_api/files/__init__.py +++ b/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: diff --git a/nc_py_api/files/files.py b/nc_py_api/files/files.py index bc9d41bc..112ef129 100644 --- a/nc_py_api/files/files.py +++ b/nc_py_api/files/files.py @@ -326,7 +326,49 @@ 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"}, @@ -334,9 +376,15 @@ def _listdir(self, user: str, path: str, properties: list[str], depth: int, excl 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) diff --git a/tests/files_test.py b/tests/files_test.py index 8d4ea10c..4e29285d 100644 --- a/tests/files_test.py +++ b/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