Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
10 changes: 9 additions & 1 deletion CHANGELOG.md
Original file line number Diff line number Diff line change
Expand Up @@ -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

Expand Down
2 changes: 1 addition & 1 deletion README.md
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down
2 changes: 1 addition & 1 deletion nc_py_api/_version.py
Original file line number Diff line number Diff line change
@@ -1,3 +1,3 @@
"""Version of nc_py_api."""

__version__ = "0.0.41"
__version__ = "0.0.42.dev0"
25 changes: 25 additions & 0 deletions nc_py_api/files/__init__.py
Original file line number Diff line number Diff line change
Expand Up @@ -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)
Expand All @@ -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:
Expand All @@ -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:
Expand Down
60 changes: 57 additions & 3 deletions nc_py_api/files/files.py
Original file line number Diff line number Diff line change
Expand Up @@ -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)
Expand Down Expand Up @@ -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)

Expand Down
46 changes: 46 additions & 0 deletions tests/files_test.py
Original file line number Diff line number Diff line change
Expand Up @@ -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