added new FileAPI: get_versions and restore_version (#108)

This was the last part of FileAPI that was missing

---------

Signed-off-by: Alexander Piskun <[email protected]>

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
 

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      |      ❌       |      ❌       |
-
-&ast;missing `File version` support.<br>
-&ast;&ast;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      |      ❌       |      ❌       |
+
+&ast;_available only for NextcloudApp_
 
 ### Differences between the Nextcloud and NextcloudApp classes
 

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."

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,23 +435,38 @@ 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("/"):
                     del result[index]
                     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:

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)