nextcloud.api_wrappers.webdav.WebDAV.set_favorites() - Code Metrics - Inspection of "Add the ability to create remote directory trees." - EnterpriseyIntranet/nextcloud-API - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Pull Request — master (#44)

by Matěj

created 2019-10-09 21:13 UTC

WebDAV.set_favorites() A

↳ Parent: nextcloud.api_wrappers.webdav

Complexity

Conditions

Size

Total Lines	19
Code Lines	4

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
eloc	4
dl	0
loc	19
rs	10
c	0
b	0
f	0
cc	1
nop	3

# -*- coding: utf-8 -*-
import re
import os

import xml.etree.ElementTree as ET

from nextcloud.base import WithRequester


class WebDAV(WithRequester):

    API_URL = "/remote.php/dav/files"

    def __init__(self, *args, **kwargs):
        super(WebDAV, self).__init__(*args)
        self.json_output = kwargs.get('json_output')

    def list_folders(self, uid, path=None, depth=1, all_properties=False):
        """
        Get path files list with files properties for given user, with given depth

        Args:
            uid (str): uid of user
            path (str/None): files path
            depth (int): depth of listing files (directories content for example)
            all_properties (bool): list all available file properties in Nextcloud

        Returns:
            list of dicts if json_output
            list of File objects if not json_output
        """
        if all_properties:
            data = """<?xml version="1.0"?>
                <d:propfind xmlns:d="DAV:" xmlns:oc="http://owncloud.org/ns"
                            xmlns:nc="http://nextcloud.org/ns">
                  <d:prop>
                        <d:getlastmodified />
                        <d:getetag />
                        <d:getcontenttype />
                        <d:resourcetype />
                        <oc:fileid />
                        <oc:permissions />
                        <oc:size />
                        <d:getcontentlength />
                        <nc:has-preview />
                        <oc:favorite />
                        <oc:comments-unread />
                        <oc:owner-display-name />
                        <oc:share-types />
                  </d:prop>
                </d:propfind>
            """
        else:
            data = None
        additional_url = uid
        if path:
            additional_url = "{}/{}".format(additional_url, path)
        resp = self.requester.propfind(additional_url=additional_url,
                                       headers={"Depth": str(depth)},
                                       data=data)
        if not resp.is_ok:
            resp.data = None
            return resp
        response_data = resp.data
        response_xml_data = ET.fromstring(response_data)
        files_data = [File(single_file) for single_file in response_xml_data]
        resp.data = files_data if not self.json_output else [each.as_dict() for each in files_data]
        return resp

    def download_file(self, uid, path):
        """
        Download file of given user by path
        File will be saved to working directory
        path argument must be valid file path

        Exception will be raised if:
            * path doesn't exist,
            * path is a directory, or if
            * file with same name already exists in working directory

        Args:
            uid (str): uid of user
            path (str): file path

        Returns:
            None
        """
        additional_url = "/".join([uid, path])
        filename = path.split('/')[-1] if '/' in path else path
        file_data = self.list_folders(uid=uid, path=path, depth=0)
        if not file_data:
            raise ValueError("Given path doesn't exist")
        file_resource_type = (file_data.data[0].get('resource_type')
                              if self.json_output
                              else file_data.data[0].resource_type)
        if file_resource_type == File.COLLECTION_RESOURCE_TYPE:
            raise ValueError("This is a collection, please specify file path")
        if filename in os.listdir('./'):
            raise ValueError("File with such name already exists in this directory")
        res = self.requester.download(additional_url)
        with open(filename, 'wb') as f:
            f.write(res.data)

    def upload_file(self, uid, local_filepath, remote_filepath):
        """
        Upload file to Nextcloud storage

        Args:
            uid (str): uid of user
            local_filepath (str): path to file on local storage
            remote_filepath (str): path where to upload file on Nextcloud storage
        """
        with open(local_filepath, 'rb') as f:
            file_contents = f.read()
        return self.upload_file_contents(uid, file_contents, remote_filepath)

    def upload_file_contents(self, uid, file_contents, remote_filepath):
        """
        Upload file to Nextcloud storage

        Args:
            uid (str): uid of user
            file_contents (bytes): Bytes the file to be uploaded consists of
            remote_filepath (str): path where to upload file on Nextcloud storage
        """
        additional_url = "/".join([uid, remote_filepath])
        return self.requester.put(additional_url, data=file_contents)

    def create_folder(self, uid, folder_path):
        """
        Create folder on Nextcloud storage

        Args:
            uid (str): uid of user
            folder_path (str): folder path
        """
        return self.requester.make_collection(additional_url="/".join([uid, folder_path]))

    def assure_folder_exists(self, uid, folder_path):
        """
        Create folder on Nextcloud storage, don't do anything if the folder already exists.
        Args:
            uid (str): uid of user
            folder_path (str): folder path
        Returns:
        """
        ret = self.create_folder(uid, folder_path)
        return True

    def assure_tree_exists(self, uid, tree_path):
        """
        Make sure that the folder structure on Nextcloud storage exists
        Args:
            uid (str): uid of user
            folder_path (str): The folder tree
        Returns:
        """
        tree = pathlib.PurePath(tree_path)

        parents = [f for f in tree.parents]
        for subf in parents[::-1]:
            ret = self.assure_folder_exists(uid, subf)
        return ret


    def delete_path(self, uid, path):
        """
        Delete file or folder with all content of given user by path

        Args:
            uid (str): uid of user
            path (str): file or folder path to delete
        """
        url = "/".join([uid, path])
        return self.requester.delete(url=url)

    def move_path(self, uid, path, destination_path, overwrite=False):
        """
        Move file or folder to destination

        Args:
            uid (str): uid of user
            path (str): file or folder path to move
            destionation_path (str): destination where to move
            overwrite (bool): allow destination path overriding
        """
        path_url = "/".join([uid, path])
        destination_path_url = "/".join([uid, destination_path])
        return self.requester.move(url=path_url,
                                   destination=destination_path_url, overwrite=overwrite)

    def copy_path(self, uid, path, destination_path, overwrite=False):
        """
        Copy file or folder to destination

        Args:
            uid (str): uid of user
            path (str): file or folder path to copy
            destionation_path (str): destination where to copy
            overwrite (bool): allow destination path overriding
        """
        path_url = "/".join([uid, path])
        destination_path_url = "/".join([uid, destination_path])
        return self.requester.copy(url=path_url,
                                   destination=destination_path_url, overwrite=overwrite)

    def set_favorites(self, uid, path):
        """
        Set files of a user favorite

        Args:
            uid (str): uid of user
            path (str): file or folder path to make favorite
        """
        data = """<?xml version="1.0"?>
        <d:propertyupdate xmlns:d="DAV:" xmlns:oc="http://owncloud.org/ns">
          <d:set>
                <d:prop>
                  <oc:favorite>1</oc:favorite>
                </d:prop>
          </d:set>
        </d:propertyupdate>
        """
        url = "/".join([uid, path])
        return self.requester.proppatch(additional_url=url, data=data)

    def list_favorites(self, uid, path=""):
        """
        Set files of a user favorite

        Args:
            uid (str): uid of user
            path (str): file or folder path to make favorite
        """
        data = """<?xml version="1.0"?>
        <oc:filter-files xmlns:d="DAV:"
                         xmlns:oc="http://owncloud.org/ns"
                         xmlns:nc="http://nextcloud.org/ns">
                 <oc:filter-rules>
                         <oc:favorite>1</oc:favorite>
                 </oc:filter-rules>
         </oc:filter-files>
        """
        url = "/".join([uid, path])
        res = self.requester.report(additional_url=url, data=data)
        if not res.is_ok:
            res.data = None
            return res
        response_xml_data = ET.fromstring(res.data)
        files_data = [File(single_file) for single_file in response_xml_data]
        res.data = files_data if not self.json_output else [each.as_dict() for each in files_data]
        return res


class File(object):

    SUCCESS_STATUS = 'HTTP/1.1 200 OK'

    # key is NextCloud property, value is python variable name
    FILE_PROPERTIES = {
        # d:
        "getlastmodified": "last_modified",
        "getetag": "etag",
        "getcontenttype": "content_type",
        "resourcetype": "resource_type",
        "getcontentlength": "content_length",
        # oc:
        "id": "id",
        "fileid": "file_id",
        "favorite": "favorite",
        "comments-href": "comments_href",
        "comments-count": "comments_count",
        "comments-unread": "comments_unread",
        "owner-id": "owner_id",
        "owner-display-name": "owner_display_name",
        "share-types": "share_types",
        "checksums": "check_sums",
        "size": "size",
        "href": "href",
        # nc:
        "has-preview": "has_preview",
    }
    xml_namespaces_map = {
        "d": "DAV:",
        "oc": "http://owncloud.org/ns",
        "nc": "http://nextcloud.org/ns"
    }
    COLLECTION_RESOURCE_TYPE = 'collection'

    def __init__(self, xml_data):
        self.href = xml_data.find('d:href', self.xml_namespaces_map).text
        for propstat in xml_data.iter('{DAV:}propstat'):
            if propstat.find('d:status', self.xml_namespaces_map).text != self.SUCCESS_STATUS:
                continue
            for file_property in propstat.find('d:prop', self.xml_namespaces_map):
                file_property_name = re.sub("{.*}", "", file_property.tag)
                if file_property_name not in self.FILE_PROPERTIES:
                    continue
                if file_property_name == 'resourcetype':
                    value = self._extract_resource_type(file_property)
                else:
                    value = file_property.text
                setattr(self, self.FILE_PROPERTIES[file_property_name], value)

    def _extract_resource_type(self, file_property):
        file_type = list(file_property)
        if file_type:
            return re.sub("{.*}", "", file_type[0].tag)
        return None

    def as_dict(self):
        return {key: value
                for key, value in self.__dict__.items()
                if key in self.FILE_PROPERTIES.values()}


class WebDAVStatusCodes(object):
    CREATED_CODE = 201
    NO_CONTENT_CODE = 204
    MULTISTATUS_CODE = 207
    ALREADY_EXISTS_CODE = 405
    PRECONDITION_FAILED_CODE = 412


1			# -- coding: utf-8 --
2			import re
3			import os
4
5			import xml.etree.ElementTree as ET
6
7			from nextcloud.base import WithRequester
8
9
10			class WebDAV(WithRequester):
11
12			API_URL = "/remote.php/dav/files"
13
14			def __init__(self, args, *kwargs):
15			super(WebDAV, self).__init__(*args)
16			self.json_output = kwargs.get('json_output')
17
18			def list_folders(self, uid, path=None, depth=1, all_properties=False):
19			"""
20			Get path files list with files properties for given user, with given depth
21
22			Args:
23			uid (str): uid of user
24			path (str/None): files path
25			depth (int): depth of listing files (directories content for example)
26			all_properties (bool): list all available file properties in Nextcloud
27
28			Returns:
29			list of dicts if json_output
30			list of File objects if not json_output
31			"""
32			if all_properties:
33			data = """<?xml version="1.0"?>
34			<d:propfind xmlns:d="DAV:" xmlns:oc="http://owncloud.org/ns"
35			xmlns:nc="http://nextcloud.org/ns">
36			<d:prop>
37			<d:getlastmodified />
38			<d:getetag />
39			<d:getcontenttype />
40			<d:resourcetype />
41			<oc:fileid />
42			<oc:permissions />
43			<oc:size />
44			<d:getcontentlength />
45			<nc:has-preview />
46			<oc:favorite />
47			<oc:comments-unread />
48			<oc:owner-display-name />
49			<oc:share-types />
50			</d:prop>
51			</d:propfind>
52			"""
53			else:
54			data = None
55			additional_url = uid
56			if path:
57			additional_url = "{}/{}".format(additional_url, path)
58			resp = self.requester.propfind(additional_url=additional_url,
59			headers={"Depth": str(depth)},
60			data=data)
61			if not resp.is_ok:
62			resp.data = None
63			return resp
64			response_data = resp.data
65			response_xml_data = ET.fromstring(response_data)
66			files_data = [File(single_file) for single_file in response_xml_data]
67			resp.data = files_data if not self.json_output else [each.as_dict() for each in files_data]
68			return resp
69
70			def download_file(self, uid, path):
71			"""
72			Download file of given user by path
73			File will be saved to working directory
74			path argument must be valid file path
75
76			Exception will be raised if:
77			* path doesn't exist,
78			* path is a directory, or if
79			* file with same name already exists in working directory
80
81			Args:
82			uid (str): uid of user
83			path (str): file path
84
85			Returns:
86			None
87			"""
88			additional_url = "/".join([uid, path])
89			filename = path.split('/')[-1] if '/' in path else path
90			file_data = self.list_folders(uid=uid, path=path, depth=0)
91			if not file_data:
92			raise ValueError("Given path doesn't exist")
93			file_resource_type = (file_data.data[0].get('resource_type')
94			if self.json_output
95			else file_data.data[0].resource_type)
96			if file_resource_type == File.COLLECTION_RESOURCE_TYPE:
97			raise ValueError("This is a collection, please specify file path")
98			if filename in os.listdir('./'):
99			raise ValueError("File with such name already exists in this directory")
100			res = self.requester.download(additional_url)
101			with open(filename, 'wb') as f:
102			f.write(res.data)
103
104			def upload_file(self, uid, local_filepath, remote_filepath):
105			"""
106			Upload file to Nextcloud storage
107
108			Args:
109			uid (str): uid of user
110			local_filepath (str): path to file on local storage
111			remote_filepath (str): path where to upload file on Nextcloud storage
112			"""
113			with open(local_filepath, 'rb') as f:
114			file_contents = f.read()
115			return self.upload_file_contents(uid, file_contents, remote_filepath)
116
117			def upload_file_contents(self, uid, file_contents, remote_filepath):
118			"""
119			Upload file to Nextcloud storage
120
121			Args:
122			uid (str): uid of user
123			file_contents (bytes): Bytes the file to be uploaded consists of
124			remote_filepath (str): path where to upload file on Nextcloud storage
125			"""
126			additional_url = "/".join([uid, remote_filepath])
127			return self.requester.put(additional_url, data=file_contents)
128
129			def create_folder(self, uid, folder_path):
130			"""
131			Create folder on Nextcloud storage
132
133			Args:
134			uid (str): uid of user
135			folder_path (str): folder path
136			"""
137			return self.requester.make_collection(additional_url="/".join([uid, folder_path]))
138
139			def assure_folder_exists(self, uid, folder_path):
140			"""
141			Create folder on Nextcloud storage, don't do anything if the folder already exists.
142			Args:
143			uid (str): uid of user
144			folder_path (str): folder path
145			Returns:
146			"""
147			ret = self.create_folder(uid, folder_path)
148			return True
149
150			def assure_tree_exists(self, uid, tree_path):
151			"""
152			Make sure that the folder structure on Nextcloud storage exists
153			Args:
154			uid (str): uid of user
155			folder_path (str): The folder tree
156			Returns:
157			"""
158			tree = pathlib.PurePath(tree_path)
			0 ignored issues – show Comprehensibility Best Practice introduced 2019-10-09 21:15 UTC by Report Bug Copy Issue Report The variable `pathlib` does not seem to be defined. Loading history...
159			parents = [f for f in tree.parents]
160			for subf in parents[::-1]:
161			ret = self.assure_folder_exists(uid, subf)
162			return ret
			0 ignored issues – show introduced 2019-10-09 21:15 UTC by Report Bug Copy Issue Report The variable `ret` does not seem to be defined in case the `for` loop on line `160` is not entered. Are you sure this can never be the case? Loading history...
163
164			def delete_path(self, uid, path):
165			"""
166			Delete file or folder with all content of given user by path
167
168			Args:
169			uid (str): uid of user
170			path (str): file or folder path to delete
171			"""
172			url = "/".join([uid, path])
173			return self.requester.delete(url=url)
174
175			def move_path(self, uid, path, destination_path, overwrite=False):
176			"""
177			Move file or folder to destination
178
179			Args:
180			uid (str): uid of user
181			path (str): file or folder path to move
182			destionation_path (str): destination where to move
183			overwrite (bool): allow destination path overriding
184			"""
185			path_url = "/".join([uid, path])
186			destination_path_url = "/".join([uid, destination_path])
187			return self.requester.move(url=path_url,
188			destination=destination_path_url, overwrite=overwrite)
189
190			def copy_path(self, uid, path, destination_path, overwrite=False):
191			"""
192			Copy file or folder to destination
193
194			Args:
195			uid (str): uid of user
196			path (str): file or folder path to copy
197			destionation_path (str): destination where to copy
198			overwrite (bool): allow destination path overriding
199			"""
200			path_url = "/".join([uid, path])
201			destination_path_url = "/".join([uid, destination_path])
202			return self.requester.copy(url=path_url,
203			destination=destination_path_url, overwrite=overwrite)
204
205			def set_favorites(self, uid, path):
206			"""
207			Set files of a user favorite
208
209			Args:
210			uid (str): uid of user
211			path (str): file or folder path to make favorite
212			"""
213			data = """<?xml version="1.0"?>
214			<d:propertyupdate xmlns:d="DAV:" xmlns:oc="http://owncloud.org/ns">
215			<d:set>
216			<d:prop>
217			<oc:favorite>1</oc:favorite>
218			</d:prop>
219			</d:set>
220			</d:propertyupdate>
221			"""
222			url = "/".join([uid, path])
223			return self.requester.proppatch(additional_url=url, data=data)
224
225			def list_favorites(self, uid, path=""):
226			"""
227			Set files of a user favorite
228
229			Args:
230			uid (str): uid of user
231			path (str): file or folder path to make favorite
232			"""
233			data = """<?xml version="1.0"?>
234			<oc:filter-files xmlns:d="DAV:"
235			xmlns:oc="http://owncloud.org/ns"
236			xmlns:nc="http://nextcloud.org/ns">
237			<oc:filter-rules>
238			<oc:favorite>1</oc:favorite>
239			</oc:filter-rules>
240			</oc:filter-files>
241			"""
242			url = "/".join([uid, path])
243			res = self.requester.report(additional_url=url, data=data)
244			if not res.is_ok:
245			res.data = None
246			return res
247			response_xml_data = ET.fromstring(res.data)
248			files_data = [File(single_file) for single_file in response_xml_data]
249			res.data = files_data if not self.json_output else [each.as_dict() for each in files_data]
250			return res
251
252
253			class File(object):
254
255			SUCCESS_STATUS = 'HTTP/1.1 200 OK'
256
257			# key is NextCloud property, value is python variable name
258			FILE_PROPERTIES = {
259			# d:
260			"getlastmodified": "last_modified",
261			"getetag": "etag",
262			"getcontenttype": "content_type",
263			"resourcetype": "resource_type",
264			"getcontentlength": "content_length",
265			# oc:
266			"id": "id",
267			"fileid": "file_id",
268			"favorite": "favorite",
269			"comments-href": "comments_href",
270			"comments-count": "comments_count",
271			"comments-unread": "comments_unread",
272			"owner-id": "owner_id",
273			"owner-display-name": "owner_display_name",
274			"share-types": "share_types",
275			"checksums": "check_sums",
276			"size": "size",
277			"href": "href",
278			# nc:
279			"has-preview": "has_preview",
280			}
281			xml_namespaces_map = {
282			"d": "DAV:",
283			"oc": "http://owncloud.org/ns",
284			"nc": "http://nextcloud.org/ns"
285			}
286			COLLECTION_RESOURCE_TYPE = 'collection'
287
288			def __init__(self, xml_data):
289			self.href = xml_data.find('d:href', self.xml_namespaces_map).text
290			for propstat in xml_data.iter('{DAV:}propstat'):
291			if propstat.find('d:status', self.xml_namespaces_map).text != self.SUCCESS_STATUS:
292			continue
293			for file_property in propstat.find('d:prop', self.xml_namespaces_map):
294			file_property_name = re.sub("{.*}", "", file_property.tag)
295			if file_property_name not in self.FILE_PROPERTIES:
296			continue
297			if file_property_name == 'resourcetype':
298			value = self._extract_resource_type(file_property)
299			else:
300			value = file_property.text
301			setattr(self, self.FILE_PROPERTIES[file_property_name], value)
302
303			def _extract_resource_type(self, file_property):
304			file_type = list(file_property)
305			if file_type:
306			return re.sub("{.*}", "", file_type[0].tag)
307			return None
308
309			def as_dict(self):
310			return {key: value
311			for key, value in self.__dict__.items()
312			if key in self.FILE_PROPERTIES.values()}
313
314
315			class WebDAVStatusCodes(object):
316			CREATED_CODE = 201
317			NO_CONTENT_CODE = 204
318			MULTISTATUS_CODE = 207
319			ALREADY_EXISTS_CODE = 405
320			PRECONDITION_FAILED_CODE = 412
321

EnterpriseyIntranet / nextcloud-API

Pull Request — master (#44)

WebDAV.set_favorites() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like