nextcloud.api_wrappers.webdav.WebDAV.get_file_property() - 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 14:13 UTC

WebDAV.get_file_property() A

↳ Parent: nextcloud.api_wrappers.webdav

Complexity

Conditions

Size

Total Lines	30
Code Lines	18

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
eloc	18
dl	0
loc	30
rs	9.0333
c	0
b	0
f	0
cc	5
nop	4

# -*- 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 'file'


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

EnterpriseyIntranet / nextcloud-API

Pull Request — master (#69)

WebDAV.get_file_property() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like