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