From f19179a5f5b5a406dd13afec6604c3aca3b48f7e Mon Sep 17 00:00:00 2001 From: Alexander Piskun <13381981+bigcat88@users.noreply.github.com> Date: Wed, 30 Aug 2023 17:09:57 +0300 Subject: [PATCH] added new FileAPI: `get_versions` and `restore_version` (#108) This was the last part of FileAPI that was missing --------- Signed-off-by: Alexander Piskun --- CHANGELOG.md | 1 + README.md | 29 ++++++------- nc_py_api/files/__init__.py | 41 +++++++++++++----- nc_py_api/files/files.py | 86 ++++++++++++++++++++++++++++++++----- tests/files_test.py | 21 ++++++++- 5 files changed, 141 insertions(+), 37 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 6b727a10..ce9e9621 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -11,6 +11,7 @@ All notable changes to this project will be documented in this file. * `trashbin_restore` * `trashbin_delete` * `trashbin_cleanup` +- File Versions API: `get_versions` and `restore_version`. ### Fixed diff --git a/README.md b/README.md index 60c944de..94d56db9 100644 --- a/README.md +++ b/README.md @@ -23,21 +23,20 @@ Python library that provides a robust and well-documented API that allows develo * **Easy**: Designed to be easy to use with excellent documentation. ### Capabilities -| **_Capability_** | Nextcloud 26 | Nextcloud 27 | Nextcloud 28 | -|-------------------|:------------:|:------------:|:------------:| -| Filesystem* | ✅ | ✅ | ✅ | -| Shares | ✅ | ✅ | ✅ | -| Users & Groups | ✅ | ✅ | ✅ | -| User status | ✅ | ✅ | ✅ | -| Weather status | ✅ | ✅ | ✅ | -| Notifications | ✅ | ✅ | ✅ | -| Nextcloud Talk | ❌ | ❌ | ❌ | -| Talk Bot API** | N/A | ✅ | ✅ | -| Text Processing** | N/A | ❌ | ❌ | -| SpeechToText** | N/A | ❌ | ❌ | - -*missing `File version` support.
-**available only for NextcloudApp +| **_Capability_** | Nextcloud 26 | Nextcloud 27 | Nextcloud 28 | +|------------------|:------------:|:------------:|:------------:| +| File System | ✅ | ✅ | ✅ | +| Shares | ✅ | ✅ | ✅ | +| Users & Groups | ✅ | ✅ | ✅ | +| User status | ✅ | ✅ | ✅ | +| Weather status | ✅ | ✅ | ✅ | +| Notifications | ✅ | ✅ | ✅ | +| Nextcloud Talk | ❌ | ❌ | ❌ | +| Talk Bot API* | N/A | ✅ | ✅ | +| Text Processing* | N/A | ❌ | ❌ | +| SpeechToText* | N/A | ❌ | ❌ | + +*_available only for NextcloudApp_ ### Differences between the Nextcloud and NextcloudApp classes diff --git a/nc_py_api/files/__init__.py b/nc_py_api/files/__init__.py index 773955f0..61037007 100644 --- a/nc_py_api/files/__init__.py +++ b/nc_py_api/files/__init__.py @@ -11,24 +11,23 @@ class FsNodeInfo: """Extra FS object attributes from Nextcloud.""" - size: int - """For directories it is size of all content in it, for files it is equal to ``size``.""" - content_length: int - """Length of file in bytes, zero for directories.""" - permissions: str - """Permissions for the object.""" - favorite: bool - """Flag indicating if the object is marked as favorite.""" fileid: int """Clear file ID without Nextcloud instance ID.""" + favorite: bool + """Flag indicating if the object is marked as favorite.""" + is_version: bool + """Flag indicating if the object is File Version representation""" _last_modified: datetime.datetime _trashbin: dict def __init__(self, **kwargs): - self.size = kwargs.get("size", 0) - self.content_length = kwargs.get("content_length", 0) - self.permissions = kwargs.get("permissions", "") + self._raw_data = { + "content_length": kwargs.get("content_length", 0), + "size": kwargs.get("size", 0), + "permissions": kwargs.get("permissions", ""), + } self.favorite = kwargs.get("favorite", False) + self.is_version = False self.fileid = kwargs.get("fileid", 0) try: self.last_modified = kwargs.get("last_modified", datetime.datetime(1970, 1, 1)) @@ -39,6 +38,21 @@ def __init__(self, **kwargs): if i in kwargs: self._trashbin[i] = kwargs[i] + @property + def content_length(self) -> int: + """Length of file in bytes, zero for directories.""" + return self._raw_data["content_length"] + + @property + def size(self) -> int: + """In the case of directories it is the size of all content, for files it is equal to ``content_length``.""" + return self._raw_data["size"] + + @property + def permissions(self) -> str: + """Permissions for the object.""" + return self._raw_data["permissions"] + @property def last_modified(self) -> datetime.datetime: """Time when the object was last modified. @@ -106,6 +120,11 @@ def is_dir(self) -> bool: return self.full_path.endswith("/") def __str__(self): + if self.info.is_version: + return ( + f"File version: `{self.name}` for FileID={self.file_id}" + f" last modified at {str(self.info.last_modified)} with {self.info.content_length} bytes size." + ) return ( f"{'Dir' if self.is_dir else 'File'}: `{self.name}` with id={self.file_id}" f" last modified at {str(self.info.last_modified)} and {self.info.permissions} permissions." diff --git a/nc_py_api/files/files.py b/nc_py_api/files/files.py index 112ef129..a898498c 100644 --- a/nc_py_api/files/files.py +++ b/nc_py_api/files/files.py @@ -1,6 +1,7 @@ """Nextcloud API for working with the file system.""" import builtins +import enum import os from io import BytesIO from json import dumps, loads @@ -15,6 +16,7 @@ from httpx import Response from .._exceptions import NextcloudException, check_error +from .._misc import require_capabilities from .._session import NcSessionBasic from . import FsNode from .sharing import _FilesSharingAPI @@ -53,6 +55,16 @@ } +class PropFindType(enum.IntEnum): + """Internal enum types for ``_listdir`` and ``_lf_parse_webdav_records`` methods.""" + + DEFAULT = 0 + TRASHBIN = 1 + FAVORITE = 2 + VERSIONS_FILEID = 3 + VERSIONS_FILE_ID = 4 + + class FilesAPI: """Class that encapsulates the file system and file sharing functionality.""" @@ -305,7 +317,7 @@ def listfav(self) -> list[FsNode]: ) request_info = f"listfav: {self._session.user}" check_error(webdav_response.status_code, request_info) - return self._lf_parse_webdav_records(webdav_response, request_info, favorite=True) + return self._lf_parse_webdav_records(webdav_response, request_info, PropFindType.FAVORITE) def setfav(self, path: Union[str, FsNode], value: Union[int, bool]) -> None: """Sets or unsets favourite flag for specific file. @@ -330,7 +342,9 @@ 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) + return self._listdir( + self._session.user, "", properties=properties, depth=1, exclude_self=False, prop_type=PropFindType.TRASHBIN + ) def trashbin_restore(self, path: Union[str, FsNode]) -> None: """Restore a file/directory from the TrashBin. @@ -366,8 +380,41 @@ def trashbin_cleanup(self) -> None: 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 get_versions(self, file_object: FsNode) -> list[FsNode]: + """Returns a list of all file versions if any.""" + require_capabilities("files.versioning", self._session.capabilities) + return self._listdir( + self._session.user, + str(file_object.info.fileid) if file_object.info.fileid else file_object.file_id, + properties=PROPFIND_PROPERTIES, + depth=1, + exclude_self=False, + prop_type=PropFindType.VERSIONS_FILEID if file_object.info.fileid else PropFindType.VERSIONS_FILE_ID, + ) + + def restore_version(self, file_object: FsNode) -> None: + """Restore a file with specified version. + + :param file_object: The **FsNode** class from :py:meth:`~nc_py_api.files.files.FilesAPI.get_versions`. + """ + require_capabilities("files.versioning", self._session.capabilities) + dest = self._session.cfg.dav_endpoint + f"/versions/{self._session.user}/restore/{file_object.name}" + headers = {"Destination": dest} + response = self._session.dav( + "MOVE", + path=f"/versions/{self._session.user}/{file_object.user_path}", + headers=headers, + ) + check_error(response.status_code, f"restore_version: user={self._session.user}, src={file_object.user_path}") + def _listdir( - self, user: str, path: str, properties: list[str], depth: int, exclude_self: bool, trashbin: bool = False + self, + user: str, + path: str, + properties: list[str], + depth: int, + exclude_self: bool, + prop_type: PropFindType = PropFindType.DEFAULT, ) -> list[FsNode]: root = ElementTree.Element( "d:propfind", @@ -376,7 +423,9 @@ def _listdir( prop = ElementTree.SubElement(root, "d:prop") for i in properties: ElementTree.SubElement(prop, i) - if trashbin: + if prop_type in (PropFindType.VERSIONS_FILEID, PropFindType.VERSIONS_FILE_ID): + dav_path = self._dav_get_obj_path(f"versions/{user}/versions", path, root_path="") + elif prop_type == PropFindType.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) @@ -386,8 +435,12 @@ def _listdir( 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) + + result = self._lf_parse_webdav_records( + webdav_response, + f"list: {user}, {path}, {properties}", + prop_type, + ) if exclude_self: for index, v in enumerate(result): if v.user_path.rstrip("/") == path.rstrip("/"): @@ -395,14 +448,25 @@ def _listdir( break return result - def _parse_records(self, fs_records: list[dict], favorite: bool): + def _parse_records(self, fs_records: list[dict], response_type: PropFindType) -> list[FsNode]: result: list[FsNode] = [] for record in fs_records: obj_full_path = unquote(record.get("d:href", "")) obj_full_path = obj_full_path.replace(self._session.cfg.dav_url_suffix, "").lstrip("/") propstat = record["d:propstat"] fs_node = self._parse_record(obj_full_path, propstat if isinstance(propstat, list) else [propstat]) - if favorite and not fs_node.file_id: + if fs_node.etag and response_type in ( + PropFindType.VERSIONS_FILE_ID, + PropFindType.VERSIONS_FILEID, + ): + fs_node.full_path = fs_node.full_path.rstrip("/") + fs_node.info.is_version = True + if response_type == PropFindType.VERSIONS_FILEID: + fs_node.info.fileid = int(fs_node.full_path.rsplit("/", 2)[-2]) + fs_node.file_id = str(fs_node.info.fileid) + else: + fs_node.file_id = fs_node.full_path.rsplit("/", 2)[-2] + if response_type == PropFindType.FAVORITE and not fs_node.file_id: _fs_node = self.by_path(fs_node.user_path) if _fs_node: _fs_node.info.favorite = True @@ -444,7 +508,9 @@ def _parse_record(full_path: str, prop_stats: list[dict]) -> FsNode: # xz = prop.get("oc:dDC", "") return FsNode(full_path, **fs_node_args) - def _lf_parse_webdav_records(self, webdav_res: Response, info: str, favorite=False) -> list[FsNode]: + def _lf_parse_webdav_records( + self, webdav_res: Response, info: str, response_type: PropFindType = PropFindType.DEFAULT + ) -> list[FsNode]: check_error(webdav_res.status_code, info=info) if webdav_res.status_code != 207: # multistatus raise NextcloudException(webdav_res.status_code, "Response is not a multistatus.", info=info) @@ -453,7 +519,7 @@ def _lf_parse_webdav_records(self, webdav_res: Response, info: str, favorite=Fal err = response_data["d:error"] raise NextcloudException(reason=f'{err["s:exception"]}: {err["s:message"]}'.replace("\n", ""), info=info) response = response_data["d:multistatus"].get("d:response", []) - return self._parse_records([response] if isinstance(response, dict) else response, favorite) + return self._parse_records([response] if isinstance(response, dict) else response, response_type) @staticmethod def _dav_get_obj_path(user: str, path: str = "", root_path="/files") -> str: diff --git a/tests/files_test.py b/tests/files_test.py index 4e29285d..7b7394de 100644 --- a/tests/files_test.py +++ b/tests/files_test.py @@ -614,7 +614,7 @@ def test_trashbin(nc): # one object now in a trashbin r = nc.files.trashbin_list() assert len(r) == 1 - # check properties types of FsNode + # check types of FsNode properties i: FsNode = r[0] assert i.info.in_trash is True assert i.info.trashbin_filename.find("nc_py_api_temp.txt") != -1 @@ -640,3 +640,22 @@ def test_trashbin(nc): # no files in trashbin r = nc.files.trashbin_list() assert not r + + +def test_file_versions(nc): + if nc.check_capabilities("files.versioning"): + pytest.skip("Need 'Versions' App to be enabled.") + for i in (0, 1): + nc.files.delete("nc_py_api_file_versions_test.txt", not_fail=True) + nc.files.upload("nc_py_api_file_versions_test.txt", content=b"22") + new_file = nc.files.upload("nc_py_api_file_versions_test.txt", content=b"333") + if i: + new_file = nc.files.by_id(new_file) + versions = nc.files.get_versions(new_file) + assert versions + version_str = str(versions[0]) + assert version_str.find("File version") != -1 + assert version_str.find("bytes size") != -1 + nc.files.restore_version(versions[0]) + assert nc.files.download(new_file) == b"22" + nc.files.delete(new_file)