FilesAPI: SystemTags implementation (#115)

Note: New `list_by_criteria` should be much faster then `listfav` --------- Signed-off-by: Alexander Piskun <bigcat88@icloud.com>
cloud-py-api · Sep 6, 2023 · e927247 · e927247
1 parent 0f85ea5
commit e927247
Show file tree

Hide file tree

Showing 8 changed files with 272 additions and 46 deletions.
diff --git a/CHANGELOG.md b/CHANGELOG.md
@@ -2,11 +2,16 @@
 
 All notable changes to this project will be documented in this file.
 
-## [0.0.44 - 2023-09-0x]
+## [0.1.0 - 2023-09-06]
 
 ### Added
 
-- Activity API: `get_filters` and `get_activities`.
+- Activity API: `get_filters` and `get_activities`. #112
+- FilesAPI: added `tags` support. #115
+
+### Changed
+
+- FilesAPI: removed `listfav` method, use new more powerful `list_by_criteria` method. #115
 
 ### Fixed
 

diff --git a/README.md b/README.md
@@ -2,7 +2,7 @@
     <img src="https://raw.githubusercontent.com/cloud-py-api/nc_py_api/main/docs/resources/nc_py_api_logo.png" width="250" alt="NcPyApi logo">
 </p>
 
-# Official Nextcloud Python Framework
+# Nextcloud Python Framework
 
 [![Analysis & Coverage](https://github.com/cloud-py-api/nc_py_api/actions/workflows/analysis-coverage.yml/badge.svg)](https://github.com/cloud-py-api/nc_py_api/actions/workflows/analysis-coverage.yml)
 [![Docs](https://github.com/cloud-py-api/nc_py_api/actions/workflows/docs.yml/badge.svg)](https://cloud-py-api.github.io/nc_py_api/)
@@ -23,18 +23,18 @@ 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 |
-|------------------|:------------:|:------------:|:------------:|
-| File System      |      ✅       |      ✅       |      ✅       |
-| Shares           |      ✅       |      ✅       |      ✅       |
-| Users & Groups   |      ✅       |      ✅       |      ✅       |
-| User status      |      ✅       |      ✅       |      ✅       |
-| Weather status   |      ✅       |      ✅       |      ✅       |
-| Notifications    |      ✅       |      ✅       |      ✅       |
-| Nextcloud Talk** |      ✅       |      ✅       |      ✅       |
-| Talk Bot API*    |     N/A      |      ✅       |      ✅       |
-| Text Processing* |     N/A      |      ❌       |      ❌       |
-| SpeechToText*    |     N/A      |      ❌       |      ❌       |
+| **_Capability_**      | Nextcloud 26 | Nextcloud 27 | Nextcloud 28 |
+|-----------------------|:------------:|:------------:|:------------:|
+| File System & Tags    |      ✅       |      ✅       |      ✅       |
+| Nextcloud Talk**      |      ✅       |      ✅       |      ✅       |
+| Notifications         |      ✅       |      ✅       |      ✅       |
+| Shares                |      ✅       |      ✅       |      ✅       |
+| Users & Groups        |      ✅       |      ✅       |      ✅       |
+| User & Weather status |      ✅       |      ✅       |      ✅       |
+| Weather status        |      ✅       |      ✅       |      ✅       |
+| Talk Bot API*         |     N/A      |      ✅       |      ✅       |
+| Text Processing*      |     N/A      |      ❌       |      ❌       |
+| SpeechToText*         |     N/A      |      ❌       |      ❌       |
 
 &ast;_available only for NextcloudApp_<br>
 &ast;&ast; _work is in progress, not all API's is described, yet._

diff --git a/docs/reference/Files/Files.rst b/docs/reference/Files/Files.rst
@@ -17,3 +17,6 @@ All File APIs are designed to work relative to the current user.
 
 .. autoclass:: nc_py_api.files.FilePermissions
     :members:
+
+.. autoclass:: nc_py_api.files.SystemTag
+    :members:
diff --git a/nc_py_api/_session.py b/nc_py_api/_session.py
@@ -229,11 +229,21 @@ def _ocs(self, method: str, path_params: str, headers: dict, data: Optional[byte
             raise NextcloudException(status_code=ocs_meta["statuscode"], reason=ocs_meta["message"], info=info)
         return response_data["ocs"]["data"]
 
-    def dav(self, method: str, path: str, data: Optional[Union[str, bytes]] = None, **kwargs) -> Response:
+    def dav(
+        self,
+        method: str,
+        path: str,
+        data: Optional[Union[str, bytes]] = None,
+        json: Optional[Union[dict, list]] = None,
+        **kwargs,
+    ) -> Response:
         headers = kwargs.pop("headers", {})
         data_bytes = None
         if data is not None:
             data_bytes = data.encode("UTF-8") if isinstance(data, str) else data
+        elif json is not None:
+            headers.update({"Content-Type": "application/json"})
+            data_bytes = dumps(json).encode("utf-8")
         return self._dav(method, quote(self.cfg.dav_url_suffix + path), headers, data_bytes, **kwargs)
 
     def dav_stream(

diff --git a/nc_py_api/files/__init__.py b/nc_py_api/files/__init__.py
@@ -208,3 +208,31 @@ class FilePermissions(enum.IntFlag):
     """Access to remove object(s)"""
     PERMISSION_SHARE = 16
     """Access to re-share object(s)"""
+
+
+@dataclasses.dataclass
+class SystemTag:
+    """Nextcloud System Tag."""
+
+    def __init__(self, raw_data: dict):
+        self._raw_data = raw_data
+
+    @property
+    def tag_id(self) -> int:
+        """Unique numeric identifier of the Tag."""
+        return int(self._raw_data["oc:id"])
+
+    @property
+    def display_name(self) -> str:
+        """The visible Tag name."""
+        return self._raw_data.get("oc:display-name", str(self.tag_id))
+
+    @property
+    def user_visible(self) -> bool:
+        """Flag indicating if the Tag is visible in the UI."""
+        return bool(self._raw_data.get("oc:user-visible", "false").lower() == "true")
+
+    @property
+    def user_assignable(self) -> bool:
+        """Flag indicating if User can assign this Tag."""
+        return bool(self._raw_data.get("oc:user-assignable", "false").lower() == "true")
diff --git a/nc_py_api/files/files.py b/nc_py_api/files/files.py
@@ -15,10 +15,10 @@
 import xmltodict
 from httpx import Response
 
-from .._exceptions import NextcloudException, check_error
-from .._misc import require_capabilities
+from .._exceptions import NextcloudException, NextcloudExceptionNotFound, check_error
+from .._misc import clear_from_params_empty, require_capabilities
 from .._session import NcSessionBasic
-from . import FsNode
+from . import FsNode, SystemTag
 from .sharing import _FilesSharingAPI
 
 PROPFIND_PROPERTIES = [
@@ -60,9 +60,8 @@ class PropFindType(enum.IntEnum):
 
     DEFAULT = 0
     TRASHBIN = 1
-    FAVORITE = 2
-    VERSIONS_FILEID = 3
-    VERSIONS_FILE_ID = 4
+    VERSIONS_FILEID = 2
+    VERSIONS_FILE_ID = 3
 
 
 class FilesAPI:
@@ -130,7 +129,7 @@ def find(self, req: list, path: Union[str, FsNode] = "") -> list[FsNode]:
         headers = {"Content-Type": "text/xml"}
         webdav_response = self._session.dav("SEARCH", "", data=self._element_tree_as_str(root), headers=headers)
         request_info = f"find: {self._session.user}, {req}, {path}"
-        return self._lf_parse_webdav_records(webdav_response, request_info)
+        return self._lf_parse_webdav_response(webdav_response, request_info)
 
     def download(self, path: Union[str, FsNode]) -> bytes:
         """Downloads and returns the content of a file.
@@ -305,20 +304,37 @@ def copy(self, path_src: Union[str, FsNode], path_dest: Union[str, FsNode], over
         check_error(response.status_code, f"copy: user={self._session.user}, src={path_src}, dest={dest}, {overwrite}")
         return self.find(req=["eq", "fileid", response.headers["OC-FileId"]])[0]
 
-    def listfav(self) -> list[FsNode]:
-        """Returns a list of the current user's favorite files."""
+    def list_by_criteria(
+        self, properties: Optional[list[str]] = None, tags: Optional[list[Union[int, SystemTag]]] = None
+    ) -> list[FsNode]:
+        """Returns a list of all files/directories for the current user filtered by the specified values.
+
+        :param properties: List of ``properties`` that should have been set for the file.
+            Supported values: **favorite**
+        :param tags: List of ``tags ids`` or ``SystemTag`` that should have been set for the file.
+        """
+        if not properties and not tags:
+            raise ValueError("Either specify 'properties' or 'tags' to filter results.")
         root = ElementTree.Element(
             "oc:filter-files",
             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 PROPFIND_PROPERTIES:
+            ElementTree.SubElement(prop, i)
         xml_filter_rules = ElementTree.SubElement(root, "oc:filter-rules")
-        ElementTree.SubElement(xml_filter_rules, "oc:favorite").text = "1"
+        if properties and "favorite" in properties:
+            ElementTree.SubElement(xml_filter_rules, "oc:favorite").text = "1"
+        if tags:
+            for v in tags:
+                tag_id = v.tag_id if isinstance(v, SystemTag) else v
+                ElementTree.SubElement(xml_filter_rules, "oc:systemtag").text = str(tag_id)
         webdav_response = self._session.dav(
             "REPORT", self._dav_get_obj_path(self._session.user), data=self._element_tree_as_str(root)
         )
-        request_info = f"listfav: {self._session.user}"
+        request_info = f"list_files_by_criteria: {self._session.user}"
         check_error(webdav_response.status_code, request_info)
-        return self._lf_parse_webdav_records(webdav_response, request_info, PropFindType.FAVORITE)
+        return self._lf_parse_webdav_response(webdav_response, request_info)
 
     def setfav(self, path: Union[str, FsNode], value: Union[int, bool]) -> None:
         """Sets or unsets favourite flag for specific file.
@@ -408,6 +424,108 @@ def restore_version(self, file_object: FsNode) -> None:
         )
         check_error(response.status_code, f"restore_version: user={self._session.user}, src={file_object.user_path}")
 
+    def list_tags(self) -> list[SystemTag]:
+        """Returns list of the avalaible Tags."""
+        root = ElementTree.Element(
+            "d:propfind",
+            attrib={"xmlns:d": "DAV:", "xmlns:oc": "http://owncloud.org/ns"},
+        )
+        properties = ["oc:id", "oc:display-name", "oc:user-visible", "oc:user-assignable"]
+        prop_element = ElementTree.SubElement(root, "d:prop")
+        for i in properties:
+            ElementTree.SubElement(prop_element, i)
+        response = self._session.dav("PROPFIND", "/systemtags", self._element_tree_as_str(root))
+        result = []
+        records = self._webdav_response_to_records(response, "list_tags")
+        for record in records:
+            prop_stat = record["d:propstat"]
+            if str(prop_stat.get("d:status", "")).find("200 OK") == -1:
+                continue
+            result.append(SystemTag(prop_stat["d:prop"]))
+        return result
+
+    def create_tag(self, name: str, user_visible: bool = True, user_assignable: bool = True) -> None:
+        """Creates a new Tag.
+
+        :param name: Name of the tag.
+        :param user_visible: Should be Tag visible in the UI.
+        :param user_assignable: Can Tag be assigned from the UI.
+        """
+        response = self._session.dav(
+            "POST",
+            path="/systemtags",
+            json={
+                "name": name,
+                "userVisible": user_visible,
+                "userAssignable": user_assignable,
+            },
+        )
+        check_error(response.status_code, info=f"create_tag({name})")
+
+    def update_tag(
+        self,
+        tag_id: Union[int, SystemTag],
+        name: Optional[str] = None,
+        user_visible: Optional[bool] = None,
+        user_assignable: Optional[bool] = None,
+    ) -> None:
+        """Updates the Tag information."""
+        tag_id = tag_id.tag_id if isinstance(tag_id, SystemTag) else tag_id
+        root = ElementTree.Element(
+            "d:propertyupdate",
+            attrib={
+                "xmlns:d": "DAV:",
+                "xmlns:oc": "http://owncloud.org/ns",
+            },
+        )
+        properties = {
+            "oc:display-name": name,
+            "oc:user-visible": "true" if user_visible is True else "false" if user_visible is False else None,
+            "oc:user-assignable": "true" if user_assignable is True else "false" if user_assignable is False else None,
+        }
+        clear_from_params_empty(list(properties.keys()), properties)
+        if not properties:
+            raise ValueError("No property specified to change.")
+        xml_set = ElementTree.SubElement(root, "d:set")
+        prop_element = ElementTree.SubElement(xml_set, "d:prop")
+        for k, v in properties.items():
+            ElementTree.SubElement(prop_element, k).text = v
+        response = self._session.dav("PROPPATCH", f"/systemtags/{tag_id}", self._element_tree_as_str(root))
+        check_error(response.status_code, info=f"update_tag({tag_id})")
+
+    def delete_tag(self, tag_id: Union[int, SystemTag]) -> None:
+        """Deletes the tag."""
+        tag_id = tag_id.tag_id if isinstance(tag_id, SystemTag) else tag_id
+        response = self._session.dav("DELETE", f"/systemtags/{tag_id}")
+        check_error(response.status_code, info=f"delete_tag({tag_id})")
+
+    def tag_by_name(self, tag_name: str) -> SystemTag:
+        """Returns Tag info by its name if found or ``None`` otherwise."""
+        r = [i for i in self.list_tags() if i.display_name == tag_name]
+        if not r:
+            raise NextcloudExceptionNotFound(f"Tag with name='{tag_name}' not found.")
+        return r[0]
+
+    def assign_tag(self, file_id: Union[FsNode, int], tag_id: Union[SystemTag, int]) -> None:
+        """Assigns Tag to a file/directory."""
+        self._file_change_tag_state(file_id, tag_id, True)
+
+    def unassign_tag(self, file_id: Union[FsNode, int], tag_id: Union[SystemTag, int]) -> None:
+        """Removes Tag from a file/directory."""
+        self._file_change_tag_state(file_id, tag_id, False)
+
+    def _file_change_tag_state(
+        self, file_id: Union[FsNode, int], tag_id: Union[SystemTag, int], tag_state: bool
+    ) -> None:
+        request = "PUT" if tag_state else "DELETE"
+        fs_object = file_id.info.fileid if isinstance(file_id, FsNode) else file_id
+        tag = tag_id.tag_id if isinstance(tag_id, SystemTag) else tag_id
+        response = self._session.dav(request, f"/systemtags-relations/files/{fs_object}/{tag}")
+        check_error(
+            response.status_code,
+            info=f"({'Adding' if tag_state else 'Removing'} `{tag}` {'to' if tag_state else 'from'} {fs_object})",
+        )
+
     def _listdir(
         self,
         user: str,
@@ -437,7 +555,7 @@ def _listdir(
             headers={"Depth": "infinity" if depth == -1 else str(depth)},
         )
 
-        result = self._lf_parse_webdav_records(
+        result = self._lf_parse_webdav_response(
             webdav_response,
             f"list: {user}, {path}, {properties}",
             prop_type,
@@ -467,12 +585,7 @@ def _parse_records(self, fs_records: list[dict], response_type: PropFindType) ->
                     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
-                    result.append(_fs_node)
-            elif fs_node.file_id:
+            if fs_node.file_id:
                 result.append(fs_node)
         return result
 
@@ -509,9 +622,13 @@ 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(
+    def _lf_parse_webdav_response(
         self, webdav_res: Response, info: str, response_type: PropFindType = PropFindType.DEFAULT
     ) -> list[FsNode]:
+        return self._parse_records(self._webdav_response_to_records(webdav_res, info), response_type)
+
+    @staticmethod
+    def _webdav_response_to_records(webdav_res: Response, info: str) -> list[dict]:
         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)
@@ -520,7 +637,7 @@ def _lf_parse_webdav_records(
             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, response_type)
+        return [response] if isinstance(response, dict) else response
 
     @staticmethod
     def _dav_get_obj_path(user: str, path: str = "", root_path="/files") -> str:

diff --git a/tests/actual_tests/conftest.py b/tests/actual_tests/conftest.py
@@ -28,16 +28,16 @@ def init_filesystem_for_user(nc_any, rand_bytes):
     /test_dir/subdir/test_empty_text.txt
     /test_dir/subdir/test_64_bytes.bin
     /test_dir/subdir/test_12345_text.txt
-    /test_dir/subdir/test_generated_image.png
+    /test_dir/subdir/test_generated_image.png       **Favorite**
     /test_dir/test_empty_child_dir/
     /test_dir/test_empty_text.txt
     /test_dir/test_64_bytes.bin
     /test_dir/test_12345_text.txt
-    /test_dir/test_generated_image.png
+    /test_dir/test_generated_image.png              **Favorite**
     /test_empty_text.txt
     /test_64_bytes.bin
     /test_12345_text.txt
-    /test_generated_image.png
+    /test_generated_image.png                       **Favorite**
     /test_dir_tmp
     """
     clean_filesystem_for_user(nc_any)
@@ -55,6 +55,7 @@ def init_folder(folder: str = ""):
         nc_any.files.upload(path.join(folder, "test_12345_text.txt"), content="12345")
         im.seek(0)
         nc_any.files.upload(path.join(folder, "test_generated_image.png"), content=im.read())
+        nc_any.files.setfav(path.join(folder, "test_generated_image.png"), True)
 
     init_folder()
     init_folder("test_dir")