From e927247d1c96dcb61028e305cdc038809ebb8454 Mon Sep 17 00:00:00 2001 From: Alexander Piskun <13381981+bigcat88@users.noreply.github.com> Date: Wed, 6 Sep 2023 21:19:19 +0300 Subject: [PATCH] FilesAPI: SystemTags implementation (#115) Note: New `list_by_criteria` should be much faster then `listfav` --------- Signed-off-by: Alexander Piskun --- CHANGELOG.md | 9 +- README.md | 26 ++--- docs/reference/Files/Files.rst | 3 + nc_py_api/_session.py | 12 ++- nc_py_api/files/__init__.py | 28 ++++++ nc_py_api/files/files.py | 159 +++++++++++++++++++++++++++---- tests/actual_tests/conftest.py | 7 +- tests/actual_tests/files_test.py | 74 ++++++++++++-- 8 files changed, 272 insertions(+), 46 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index e9f9ca17..de9b0339 100644 --- 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 index 726e7f25..9ec5ae2f 100644 --- a/README.md +++ b/README.md @@ -2,7 +2,7 @@ NcPyApi logo

-# 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 | ❌ | ❌ | *_available only for NextcloudApp_
** _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 index c0c44d1a..a91d0cc8 100644 --- 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 index aa440f73..3ad92773 100644 --- 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 index 61037007..69c024df 100644 --- 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 index 1442492c..2f1d08f4 100644 --- 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 index 51be8612..3dde468d 100644 --- 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") diff --git a/tests/actual_tests/files_test.py b/tests/actual_tests/files_test.py index 5ae64eb4..7c47b2b8 100644 --- a/tests/actual_tests/files_test.py +++ b/tests/actual_tests/files_test.py @@ -1,3 +1,4 @@ +import contextlib import math import os import zipfile @@ -10,7 +11,7 @@ import pytest -from nc_py_api import FsNode, NextcloudException +from nc_py_api import FsNode, NextcloudException, NextcloudExceptionNotFound class MyBytesIO(BytesIO): @@ -256,20 +257,26 @@ def test_mkdir_delete_with_end_slash(nc_any): def test_favorites(nc_any): - favorites = nc_any.files.listfav() + favorites = nc_any.files.list_by_criteria(["favorite"]) + favorites = [i for i in favorites if i.name != "test_generated_image.png"] for favorite in favorites: nc_any.files.setfav(favorite.user_path, False) - assert not nc_any.files.listfav() + favorites = nc_any.files.list_by_criteria(["favorite"]) + favorites = [i for i in favorites if i.name != "test_generated_image.png"] + assert not favorites files = ("test_dir_tmp/fav1.txt", "test_dir_tmp/fav2.txt", "test_dir_tmp/fav3.txt") for n in files: nc_any.files.upload(n, content=n) nc_any.files.setfav(n, True) - favorites = nc_any.files.listfav() + favorites = nc_any.files.list_by_criteria(["favorite"]) + favorites = [i for i in favorites if i.name != "test_generated_image.png"] assert len(favorites) == 3 for favorite in favorites: assert isinstance(favorite, FsNode) nc_any.files.setfav(favorite, False) - assert len(nc_any.files.listfav()) == 0 + favorites = nc_any.files.list_by_criteria(["favorite"]) + favorites = [i for i in favorites if i.name != "test_generated_image.png"] + assert not favorites def test_copy_file(nc_any, rand_bytes): @@ -366,7 +373,6 @@ def test_fs_node_fields(nc_any): assert len(results) == 6 for _, result in enumerate(results): assert result.user == "admin" - assert not result.info.favorite if result.name == "subdir": assert result.user_path == "test_dir/subdir/" assert result.is_dir @@ -374,6 +380,7 @@ def test_fs_node_fields(nc_any): assert result.info.size == 2364 assert result.info.content_length == 0 assert result.info.permissions == "RGDNVCK" + assert result.info.favorite is False elif result.name == "test_empty_child_dir": assert result.user_path == "test_dir/test_empty_child_dir/" assert result.is_dir @@ -381,6 +388,7 @@ def test_fs_node_fields(nc_any): assert result.info.size == 0 assert result.info.content_length == 0 assert result.info.permissions == "RGDNVCK" + assert result.info.favorite is False elif result.name == "test_generated_image.png": assert result.user_path == "test_dir/test_generated_image.png" assert not result.is_dir @@ -388,6 +396,7 @@ def test_fs_node_fields(nc_any): assert result.info.size > 900 assert result.info.size == result.info.content_length assert result.info.permissions == "RGDNVW" + assert result.info.favorite is True elif result.name == "test_empty_text.txt": assert result.user_path == "test_dir/test_empty_text.txt" assert not result.is_dir @@ -395,6 +404,7 @@ def test_fs_node_fields(nc_any): assert not result.info.size assert not result.info.content_length assert result.info.permissions == "RGDNVW" + assert result.info.favorite is False res_by_id = nc_any.files.by_id(result.file_id) assert res_by_id @@ -554,3 +564,55 @@ def test_file_versions(nc_any): assert version_str.find("bytes size") != -1 nc_any.files.restore_version(versions[0]) assert nc_any.files.download(new_file) == b"22" + + +def test_create_update_delete_tag(nc_any): + with contextlib.suppress(NextcloudExceptionNotFound): + nc_any.files.delete_tag(nc_any.files.tag_by_name("test_nc_py_api")) + with contextlib.suppress(NextcloudExceptionNotFound): + nc_any.files.delete_tag(nc_any.files.tag_by_name("test_nc_py_api2")) + nc_any.files.create_tag("test_nc_py_api", True, True) + tag = nc_any.files.tag_by_name("test_nc_py_api") + assert isinstance(tag.tag_id, int) + assert tag.display_name == "test_nc_py_api" + assert tag.user_visible is True + assert tag.user_assignable is True + nc_any.files.update_tag(tag, "test_nc_py_api2", False, False) + with pytest.raises(NextcloudExceptionNotFound): + nc_any.files.tag_by_name("test_nc_py_api") + tag = nc_any.files.tag_by_name("test_nc_py_api2") + assert tag.display_name == "test_nc_py_api2" + assert tag.user_visible is False + assert tag.user_assignable is False + nc_any.files.delete_tag(tag) + with pytest.raises(ValueError): + nc_any.files.update_tag(tag) + + +def test_assign_unassign_tag(nc_any): + with contextlib.suppress(NextcloudExceptionNotFound): + nc_any.files.delete_tag(nc_any.files.tag_by_name("test_nc_py_api")) + with contextlib.suppress(NextcloudExceptionNotFound): + nc_any.files.delete_tag(nc_any.files.tag_by_name("test_nc_py_api2")) + nc_any.files.create_tag("test_nc_py_api", True, False) + nc_any.files.create_tag("test_nc_py_api2", False, False) + tag1 = nc_any.files.tag_by_name("test_nc_py_api") + assert tag1.user_visible is True + assert tag1.user_assignable is False + tag2 = nc_any.files.tag_by_name("test_nc_py_api2") + assert tag2.user_visible is False + assert tag2.user_assignable is False + new_file = nc_any.files.upload("/test_dir_tmp/tag_test.txt", content=b"") + new_file = nc_any.files.by_id(new_file) + assert len(nc_any.files.list_by_criteria(tags=[tag1])) == 0 + nc_any.files.assign_tag(new_file, tag1) + assert len(nc_any.files.list_by_criteria(tags=[tag1])) == 1 + assert len(nc_any.files.list_by_criteria(["favorite"], tags=[tag1])) == 0 + assert len(nc_any.files.list_by_criteria(tags=[tag1, tag2.tag_id])) == 0 + nc_any.files.assign_tag(new_file, tag2.tag_id) + assert len(nc_any.files.list_by_criteria(tags=[tag1, tag2.tag_id])) == 1 + nc_any.files.unassign_tag(new_file, tag1) + assert len(nc_any.files.list_by_criteria(tags=[tag1])) == 0 + nc_any.files.assign_tag(new_file, tag1) + with pytest.raises(ValueError): + nc_any.files.list_by_criteria()