nextcloud.api_wrappers.webdav.File.__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 13:34 UTC

nextcloud.api_wrappers.webdav.File.init() B

↳ Parent: Project

Complexity

Conditions

Size

Total Lines	14
Code Lines	13

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
eloc	13
dl	0
loc	14
rs	8.6666
c	0
b	0
f	0
cc	6
nop	2

# -*- 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)


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

EnterpriseyIntranet / nextcloud-API

Pull Request — master (#69)

nextcloud.api_wrappers.webdav.File.__init__() B

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like

nextcloud.api_wrappers.webdav.File.init() B