NextCloud.api_wrappers.webdav.WebDAV.list_folders() - Code Metrics - Inspection of "Merge pull request #24 from danil-topc/webdav_api" - EnterpriseyIntranet/nextcloud-API - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Push — master ( 16f953...0e3ad4 )

by Matěj

created 2019-02-04 21:57 UTC

WebDAV.list_folders() A

↳ Parent: NextCloud.api_wrappers.webdav

Complexity

Conditions

Size

Total Lines	46
Code Lines	16

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
cc	5
eloc	16
nop	5
dl	0
loc	46
rs	9.1333
c	0
b	0
f	0

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 resp.raw.status_code != 207:
            return []
        response_data = resp.data
        response_xml_data = ET.fromstring(response_data)
        files_data = [File(single_file) for single_file in response_xml_data]
        return files_data if not self.json_output else [each.as_dict() for each in files_data]

    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
            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[0].get('resource_type') if self.json_output else file_data.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
        Returns:
        """
        with open(local_filepath, 'rb') as f:
            file_content = f.read()
        additional_url = "/".join([uid, remote_filepath])
        return self.requester.put(additional_url, data=file_content)

    def create_folder(self, uid, folder_path):
        """
        Create folder on Nextcloud storage
        Args:
            uid (str): uid of user
            folder_path (str): folder path
        Returns:
        """
        return self.requester.make_collection(additional_url="/".join([uid, folder_path]))

    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
        Returns:
        """
        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
        Returns:
        """
        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
        Returns:
        """
        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
        Returns:
        """
        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
        Returns:
        """
        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)
        response_xml_data = ET.fromstring(res.data)
        files_data = [File(single_file) for single_file in response_xml_data]
        return files_data if not self.json_output else [each.as_dict() for each in files_data]


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()}



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

EnterpriseyIntranet / nextcloud-API

Push — master ( 16f953...0e3ad4 )

WebDAV.list_folders() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like