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-03 16:51 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)


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

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

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