nextcloud.api_wrappers.webdav.WebDAV.__init__() - Code Metrics - Inspection of "Refactoring for modulable wrappers / Python2 suppo..." - EnterpriseyIntranet/nextcloud-API - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Pull Request — master (#69)

unknown

created 2021-05-04 15:11 UTC

nextcloud.api_wrappers.webdav.WebDAV.init() A

↳ Parent: nextcloud.api_wrappers.webdav

Complexity

Conditions

Size

Total Lines	3
Code Lines	3

Duplication

Lines	0
Ratio	0 %

Importance

Changes

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

# -*- coding: utf-8 -*-
"""
WebDav API wrapper
See https://doc.owncloud.com/server/developer_manual/webdav_api/tags.html
"""
import re
import os
try:
    import pathlib
except:
    import pathlib2 as pathlib

import xml.etree.ElementTree as ET
from datetime import datetime
from nextcloud.base import WebDAVApiWrapper
from nextcloud.common.collections import PropertySet
from nextcloud.common.properties import Property as Prop, NAMESPACES_MAP
from nextcloud.common.value_parsing import timestamp_to_epoch_time


class File(PropertySet):
    _attrs = [
        Prop('d:getlastmodified'),
        Prop('d:getetag'),
        Prop('d:getcontenttype'),
        Prop('d:resourcetype', parse_xml_value=(
            lambda p: File._extract_resource_type(p))),
        Prop('d:getcontentlength'),
        Prop('oc:id'),
        Prop('oc:fileid'),
        Prop('oc:favorite'),
        Prop('oc:comments-href'),
        Prop('oc:comments-count'),
        Prop('oc:comments-unread'),
        Prop('oc:owner-id'),
        Prop('oc:owner-display-name'),
        Prop('oc:share-types'),
        Prop('oc:checksums'),
        Prop('oc:size'),
        Prop('oc:href'),
        Prop('nc:has-preview')
    ]

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

    def isfile(self):
        """ say if the file is a file /!\\ ressourcetype property shall be loaded """
        return not self.resource_type

    def isdir(self):
        """ say if the file is a directory /!\\ ressourcetype property shall be loaded """
        return self.resource_type == self.COLLECTION_RESOURCE_TYPE


class WebDAV(WebDAVApiWrapper):
    """ WebDav API wrapper """
    API_URL = "/remote.php/dav/files"

    def _get_path(self, path):
        if path:
            return '/'.join([self.client.user, path]).replace('//', '/')
        else:
            return self.client.user

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

        Args:
            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
            fields (str list): file properties to fetch

        Returns:
            list of dicts if json_output
            list of File objects if not json_output
        """
        data = File.build_xml_propfind(
            use_default=all_properties,
            fields=fields
        ) if (fields or all_properties) else None
        resp = self.requester.propfind(additional_url=self._get_path(path),
                                       headers={'Depth': str(depth)},
                                       data=data)
        return File.from_response(resp, json_output=(self.json_output))

    def download_file(self, path):
        """
        Download file by path (for current user)
        File will be saved to working directory
        path argument must be valid file path
        Modified time of saved file will be synced with the file properties in Nextcloud

        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:
            path (str): file path

        Returns:
            None
        """
        filename = path.split('/')[(-1)] if '/' in path else path
        file_data = self.list_folders(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(self._get_path(path))
        with open(filename, 'wb') as f:
            f.write(res.data)

        # get timestamp of downloaded file from file property on Nextcloud
        # If it succeeded, set the timestamp to saved local file
        # If the timestamp string is invalid or broken, the timestamp is downloaded time.
        file_timestamp_str = (file_data.data[0].get('last_modified')
                              if self.json_output
                              else file_data.data[0].last_modified)
        file_timestamp = timestamp_to_epoch_time(file_timestamp_str)
        if isinstance(file_timestamp, int):
            os.utime(filename, (datetime.now().timestamp(), file_timestamp))

    def upload_file(self, local_filepath, remote_filepath, timestamp=None):
        """
        Upload file to Nextcloud storage

        Args:
            local_filepath (str): path to file on local storage
            remote_filepath (str): path where to upload file on Nextcloud storage
            timestamp (int): timestamp of upload file. If None, get time by local file.

        Returns:
            requester response
        """
        with open(local_filepath, 'rb') as f:
            file_contents = f.read()
        if timestamp is None:
            timestamp = int(os.path.getmtime(local_filepath))
        return self.upload_file_contents(file_contents, remote_filepath, timestamp)

    def upload_file_contents(self, file_contents, remote_filepath, timestamp=None):
        """
        Upload file to Nextcloud storage

        Args:
            file_contents (bytes): Bytes the file to be uploaded consists of
            remote_filepath (str): path where to upload file on Nextcloud storage
            timestamp (int):  mtime of upload file

        Returns:
            requester response
        """
        return self.requester.put_with_timestamp((self._get_path(remote_filepath)), data=file_contents,
                                                 timestamp=timestamp)

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

        Args:
            folder_path (str): folder path

        Returns:
            requester response
        """
        return self.requester.make_collection(additional_url=(self._get_path(folder_path)))

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

    def assure_tree_exists(self, tree_path):
        """
        Make sure that the folder structure on Nextcloud storage exists
        Args:
            folder_path (str): The folder tree
        Returns:
            requester response
        """
        tree = pathlib.PurePath(tree_path)
        parents = list(tree.parents)
        ret = True
        subfolders = parents[:-1][::-1] + [tree]
        for subf in subfolders:
            ret = self.assure_folder_exists(str(subf))

        return ret

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

        Args:
            path (str): file or folder path to delete

        Returns:
            requester response
        """
        return self.requester.delete(url=self._get_path(path))

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

        Args:
            path (str): file or folder path to move
            destionation_path (str): destination where to move
            overwrite (bool): allow destination path overriding

        Returns:
            requester response
        """
        return self.requester.move(url=self._get_path(path),
                                   destination=self._get_path(destination_path),
                                   overwrite=overwrite)

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

        Args:
            path (str): file or folder path to copy
            destionation_path (str): destination where to copy
            overwrite (bool): allow destination path overriding

        Returns:
            requester response
        """
        return self.requester.copy(url=self._get_path(path),
                                   destination=self._get_path(destination_path),
                                   overwrite=overwrite)

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

        Args:
            path (str): file or folder path to make favorite

        Returns:
            requester response
        """
        data = File.build_xml_propupdate({'oc': {'favorite': 1}})
        return self.requester.proppatch(additional_url=self._get_path(path), data=data)

    def list_favorites(self, path=''):
        """
        List favorites (files) of the user

        Args:
            path (str): file or folder path to make favorite

        Returns:
            requester response with <list>File in data
        """
        data = File.build_xml_propfind(
            instr='oc:filter-files', filter_rules={'oc': {'favorite': 1}})
        resp = self.requester.report(additional_url=self._get_path(path), data=data)
        return File.from_response(resp, json_output=self.json_output)

    def get_file_property(self, path, field, tag='oc'):
        """
        Fetch asked properties from a file path.

        Args:
            path (str): file or folder path to make favorite
            field (str): field name

        Returns:
            requester response with asked value in data
        """
        if ':' in field:
            tag, field = field.split(':')
        get_file_prop_xpath = '{DAV:}propstat/d:prop/%s:%s' % (tag, field)
        data = File.build_xml_propfind(fields={tag: [field]})
        resp = self.requester.propfind(additional_url=(self._get_path(path)), headers={'Depth': str(0)},
                                       data=data)
        response_data = resp.data
        resp.data = None
        if not resp.is_ok:
            return resp

        response_xml_data = ET.fromstring(response_data)
        for xml_data in response_xml_data:
            for prop in xml_data.findall(get_file_prop_xpath,
                                         NAMESPACES_MAP):
                resp.data = prop.text
            break

        return resp


1			# -- coding: utf-8 --
2			"""
3			WebDav API wrapper
4			See https://doc.owncloud.com/server/developer_manual/webdav_api/tags.html
5			"""
6			import re
7			import os
8			try:
9			import pathlib
10			except:
11			import pathlib2 as pathlib
12
13			import xml.etree.ElementTree as ET
14			from datetime import datetime
15			from nextcloud.base import WebDAVApiWrapper
16			from nextcloud.common.collections import PropertySet
17			from nextcloud.common.properties import Property as Prop, NAMESPACES_MAP
18			from nextcloud.common.value_parsing import timestamp_to_epoch_time
19
20
21			class File(PropertySet):
22			_attrs = [
23			Prop('d:getlastmodified'),
24			Prop('d:getetag'),
25			Prop('d:getcontenttype'),
26			Prop('d:resourcetype', parse_xml_value=(
27			lambda p: File._extract_resource_type(p))),
28			Prop('d:getcontentlength'),
29			Prop('oc:id'),
30			Prop('oc:fileid'),
31			Prop('oc:favorite'),
32			Prop('oc:comments-href'),
33			Prop('oc:comments-count'),
34			Prop('oc:comments-unread'),
35			Prop('oc:owner-id'),
36			Prop('oc:owner-display-name'),
37			Prop('oc:share-types'),
38			Prop('oc:checksums'),
39			Prop('oc:size'),
40			Prop('oc:href'),
41			Prop('nc:has-preview')
42			]
43
44			@staticmethod
45			def _extract_resource_type(file_property):
46			file_type = list(file_property)
47			if file_type:
48			return re.sub('{.*}', '', file_type[0].tag)
49			return None
50
51			def isfile(self):
52			""" say if the file is a file /!\\ ressourcetype property shall be loaded """
53			return not self.resource_type
54
55			def isdir(self):
56			""" say if the file is a directory /!\\ ressourcetype property shall be loaded """
57			return self.resource_type == self.COLLECTION_RESOURCE_TYPE
58
59
60			class WebDAV(WebDAVApiWrapper):
61			""" WebDav API wrapper """
62			API_URL = "/remote.php/dav/files"
63
64			def _get_path(self, path):
65			if path:
66			return '/'.join([self.client.user, path]).replace('//', '/')
67			else:
68			return self.client.user
69
70			def list_folders(self, path=None, depth=1, all_properties=False,
71			fields=None):
72			"""
73			Get path files list with files properties with given depth
74			(for current user)
75
76			Args:
77			path (str/None): files path
78			depth (int): depth of listing files (directories content for example)
79			all_properties (bool): list all available file properties in Nextcloud
80			fields (str list): file properties to fetch
81
82			Returns:
83			list of dicts if json_output
84			list of File objects if not json_output
85			"""
86			data = File.build_xml_propfind(
87			use_default=all_properties,
88			fields=fields
89			) if (fields or all_properties) else None
90			resp = self.requester.propfind(additional_url=self._get_path(path),
91			headers={'Depth': str(depth)},
92			data=data)
93			return File.from_response(resp, json_output=(self.json_output))
94
95			def download_file(self, path):
96			"""
97			Download file by path (for current user)
98			File will be saved to working directory
99			path argument must be valid file path
100			Modified time of saved file will be synced with the file properties in Nextcloud
101
102			Exception will be raised if:
103			* path doesn't exist,
104			* path is a directory, or if
105			* file with same name already exists in working directory
106
107			Args:
108			path (str): file path
109
110			Returns:
111			None
112			"""
113			filename = path.split('/')[(-1)] if '/' in path else path
114			file_data = self.list_folders(path=path, depth=0)
115			if not file_data:
116			raise ValueError("Given path doesn't exist")
117			file_resource_type = (file_data.data[0].get('resource_type')
118			if self.json_output
119			else file_data.data[0].resource_type)
120			if file_resource_type == File.COLLECTION_RESOURCE_TYPE:
121			raise ValueError("This is a collection, please specify file path")
122			if filename in os.listdir('./'):
123			raise ValueError( "File with such name already exists in this directory")
124			res = self.requester.download(self._get_path(path))
125			with open(filename, 'wb') as f:
126			f.write(res.data)
127
128			# get timestamp of downloaded file from file property on Nextcloud
129			# If it succeeded, set the timestamp to saved local file
130			# If the timestamp string is invalid or broken, the timestamp is downloaded time.
131			file_timestamp_str = (file_data.data[0].get('last_modified')
132			if self.json_output
133			else file_data.data[0].last_modified)
134			file_timestamp = timestamp_to_epoch_time(file_timestamp_str)
135			if isinstance(file_timestamp, int):
136			os.utime(filename, (datetime.now().timestamp(), file_timestamp))
137
138			def upload_file(self, local_filepath, remote_filepath, timestamp=None):
139			"""
140			Upload file to Nextcloud storage
141
142			Args:
143			local_filepath (str): path to file on local storage
144			remote_filepath (str): path where to upload file on Nextcloud storage
145			timestamp (int): timestamp of upload file. If None, get time by local file.
146
147			Returns:
148			requester response
149			"""
150			with open(local_filepath, 'rb') as f:
151			file_contents = f.read()
152			if timestamp is None:
153			timestamp = int(os.path.getmtime(local_filepath))
154			return self.upload_file_contents(file_contents, remote_filepath, timestamp)
155
156			def upload_file_contents(self, file_contents, remote_filepath, timestamp=None):
157			"""
158			Upload file to Nextcloud storage
159
160			Args:
161			file_contents (bytes): Bytes the file to be uploaded consists of
162			remote_filepath (str): path where to upload file on Nextcloud storage
163			timestamp (int): mtime of upload file
164
165			Returns:
166			requester response
167			"""
168			return self.requester.put_with_timestamp((self._get_path(remote_filepath)), data=file_contents,
169			timestamp=timestamp)
170
171			def create_folder(self, folder_path):
172			"""
173			Create folder on Nextcloud storage
174
175			Args:
176			folder_path (str): folder path
177
178			Returns:
179			requester response
180			"""
181			return self.requester.make_collection(additional_url=(self._get_path(folder_path)))
182
183			def assure_folder_exists(self, folder_path):
184			"""
185			Create folder on Nextcloud storage, don't do anything if the folder already exists.
186			Args:
187			folder_path (str): folder path
188			Returns:
189			requester response
190			"""
191			self.create_folder(folder_path)
192			return True
193
194			def assure_tree_exists(self, tree_path):
195			"""
196			Make sure that the folder structure on Nextcloud storage exists
197			Args:
198			folder_path (str): The folder tree
199			Returns:
200			requester response
201			"""
202			tree = pathlib.PurePath(tree_path)
203			parents = list(tree.parents)
204			ret = True
205			subfolders = parents[:-1][::-1] + [tree]
206			for subf in subfolders:
207			ret = self.assure_folder_exists(str(subf))
208
209			return ret
210
211			def delete_path(self, path):
212			"""
213			Delete file or folder with all content of given user by path
214
215			Args:
216			path (str): file or folder path to delete
217
218			Returns:
219			requester response
220			"""
221			return self.requester.delete(url=self._get_path(path))
222
223			def move_path(self, path, destination_path, overwrite=False):
224			"""
225			Move file or folder to destination
226
227			Args:
228			path (str): file or folder path to move
229			destionation_path (str): destination where to move
230			overwrite (bool): allow destination path overriding
231
232			Returns:
233			requester response
234			"""
235			return self.requester.move(url=self._get_path(path),
236			destination=self._get_path(destination_path),
237			overwrite=overwrite)
238
239			def copy_path(self, path, destination_path, overwrite=False):
240			"""
241			Copy file or folder to destination
242
243			Args:
244			path (str): file or folder path to copy
245			destionation_path (str): destination where to copy
246			overwrite (bool): allow destination path overriding
247
248			Returns:
249			requester response
250			"""
251			return self.requester.copy(url=self._get_path(path),
252			destination=self._get_path(destination_path),
253			overwrite=overwrite)
254
255			def set_favorites(self, path):
256			"""
257			Set files of a user favorite
258
259			Args:
260			path (str): file or folder path to make favorite
261
262			Returns:
263			requester response
264			"""
265			data = File.build_xml_propupdate({'oc': {'favorite': 1}})
266			return self.requester.proppatch(additional_url=self._get_path(path), data=data)
267
268			def list_favorites(self, path=''):
269			"""
270			List favorites (files) of the user
271
272			Args:
273			path (str): file or folder path to make favorite
274
275			Returns:
276			requester response with <list>File in data
277			"""
278			data = File.build_xml_propfind(
279			instr='oc:filter-files', filter_rules={'oc': {'favorite': 1}})
280			resp = self.requester.report(additional_url=self._get_path(path), data=data)
281			return File.from_response(resp, json_output=self.json_output)
282
283			def get_file_property(self, path, field, tag='oc'):
284			"""
285			Fetch asked properties from a file path.
286
287			Args:
288			path (str): file or folder path to make favorite
289			field (str): field name
290
291			Returns:
292			requester response with asked value in data
293			"""
294			if ':' in field:
295			tag, field = field.split(':')
296			get_file_prop_xpath = '{DAV:}propstat/d:prop/%s:%s' % (tag, field)
297			data = File.build_xml_propfind(fields={tag: [field]})
298			resp = self.requester.propfind(additional_url=(self._get_path(path)), headers={'Depth': str(0)},
299			data=data)
300			response_data = resp.data
301			resp.data = None
302			if not resp.is_ok:
303			return resp
304
305			response_xml_data = ET.fromstring(response_data)
306			for xml_data in response_xml_data:
307			for prop in xml_data.findall(get_file_prop_xpath,
308			NAMESPACES_MAP):
309			resp.data = prop.text
310			break
311
312			return resp
313

EnterpriseyIntranet / nextcloud-API

Pull Request — master (#69)

nextcloud.api_wrappers.webdav.WebDAV.__init__() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like

nextcloud.api_wrappers.webdav.WebDAV.init() A