nextcloud.api_wrappers.webdav.WebDAV.upload_file_contents() - 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-05 15:49 UTC

WebDAV.upload_file_contents() A

↳ Parent: nextcloud.api_wrappers.webdav

Complexity

Conditions

Size

Total Lines	14
Code Lines	3

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
eloc	3
dl	0
loc	14
rs	10
c	0
b	0
f	0
cc	1
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 ImportError:
    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,
    datetime_to_timestamp
)


class NextCloudFileConflict(Exception):
    """ Exception to raise when you try to create a File that alreay exists """


class File(PropertySet):
    """
    Define properties on a WebDav file/folder

    Additionnally, provide an objective CRUD API
    (that probably consume more energy than fetching specific attributes)

    Example :
    >>> root = nxc.get_folder()  # get root
    >>> def _list_rec(d, indent=""):
    >>>     # list files recursively
    >>>     print("%s%s%s" % (indent, d.basename(), '/' if d.isdir() else ''))
    >>>     if d.isdir():
    >>>         for i in d.list():
    >>>             _list_rec(i, indent=indent+"  ")
    >>>
    >>> _list_rec(root)
    """

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

    _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')
    ]

    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

    def get_relative_path(self):
        """ get path relative to user root """
        return self._wrapper.get_relative_path(self.href)

    def _get_remote_path(self, path=None):
        _url = self.get_relative_path()
        return '/'.join([_url, path]) if path else _url

    def basename(self):
        """ basename """
        _path = self._get_remote_path()
        return _path.split('/')[-2] if _path.endswith('/') else _path.split('/')[-1]

    def dirname(self):
        """ dirname """
        _path = self._get_remote_path()
        return '/'.join(_path.split('/')[:-2]) if _path.endswith('/') else '/'.join(_path.split('/')[:-1])

    def __eq__(self, b):
        return self.href == b.href

    # MINIMAL SET OF CRUD OPERATIONS
    def get_folder(self, path=None):
        """
        Get folder (see WebDav wrapper)
        :param path: if empty list current dir
        :returns: a folder (File object)

        Note : To check if sub folder exists, use get_file method
        """
        return self._wrapper.get_folder(self._get_remote_path(path))

    def get_file(self, path=None):
        """
        Get file (see WebDav wrapper)
        :param path: if empty list current dir
        :returns: a file or folder (File object)
        """
        return self._wrapper.get_file(self._get_remote_path(path))

    def list(self, subpath='', filter_rules=None):
        """
        List folder (see WebDav wrapper)
        :param subpath: if empty list current dir
        :returns: list of Files
        """
        if filter_rules:
            resp = self._wrapper.fetch_files_with_filter(
                path=self._get_remote_path(subpath),
                filter_rules=filter_rules
            )
        else:
            resp = self._wrapper.list_folders(
                self._get_remote_path(subpath),
                depth=1,
                all_properties=True
            )
        if resp.is_ok and resp.data:
            _dirs = resp.data
            # remove current dir
            if _dirs[0] == self:
                _dirs = _dirs[1:]
            return _dirs
        return []

    def upload_file(self, local_filepath, name, timestamp=None):
        """
        Upload file (see WebDav wrapper)
        :param name: name of the new file
        :returns: True if success
        """
        resp = self._wrapper.upload_file(local_filepath,
                                         self._get_remote_path(name),
                                         timestamp=timestamp)
        return resp.is_ok

    def download(self, name=None, target_dir=None):
        """
         file (see WebDav wrapper)
        :param name: name of the new file
        :returns: True if success
        """
        path = self._get_remote_path(name)
        target_path, _file_info = self._wrapper.download_file(path,
                                                              target_dir=target_dir)
        assert os.path.isfile(target_path), "Download failed"
        return target_path

    def delete(self, subpath=''):
        """
        Delete file or folder (see WebDav wrapper)
        :param subpath: if empty, delete current file
        :returns: True if success
        """
        resp = self._wrapper.delete_path(self._get_remote_path(subpath))
        return resp.is_ok


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('//', '/')
        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,
                                  wrapper=self)

    def download_file(self, path, target_dir=None):
        """
        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:
            a tuple (target_path, File object)
        """
        if not target_dir:
            target_dir = './'
        filename = path.split('/')[(-1)] if '/' in path else path
        file_data = self.get_file(path)
        if not file_data:
            raise ValueError("Given path doesn't exist")
        file_resource_type = 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(target_dir):
            raise ValueError(
                "File with such name already exists in this directory")
        filename = os.path.join(target_dir, filename)
        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.last_modified
        file_timestamp = timestamp_to_epoch_time(file_timestamp_str)
        if isinstance(file_timestamp, int):
            os.utime(filename, (
                datetime_to_timestamp(datetime.now()),
                file_timestamp))
        return (filename, file_data)

    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_file_property(self, path, update_rules):
        """
        Set file property

        Args:
            path (str): file or folder path to make favorite
            update_rules : a dict { namespace: {key : value } }

        Returns:
            requester response with list<File> in data

        Note :
            check keys in nextcloud.common.properties.NAMESPACES_MAP for namespace codes
            check object property xml_name for property name
        """
        data = File.build_xml_propupdate(update_rules)
        return self.requester.proppatch(additional_url=self._get_path(path), data=data)

    def fetch_files_with_filter(self, path='', filter_rules=''):
        """
        List files according to a filter

        Args:
            path (str): file or folder path to search
            filter_rules : a dict { namespace: {key : value } }

        Returns:
            requester response with list<File> in data

        Note :
            check keys in nextcloud.common.properties.NAMESPACES_MAP for namespace codes
            check object property xml_name for property name
        """
        data = File.build_xml_propfind(
            instr='oc:filter-files', filter_rules=filter_rules)
        resp = self.requester.report(
            additional_url=self._get_path(path), data=data)
        return File.from_response(resp, json_output=self.json_output,
                                  wrapper=self)

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

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

        Returns:
            requester response
        """
        return self.set_file_property(path, {'oc': {'favorite': 1}})

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

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

        Returns:
            requester response with list<File> in data
        """
        return self.fetch_files_with_filter(path, {'oc': {'favorite': 1}})

    def get_file_property(self, path, field, ns='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:
            ns, field = field.split(':')
        get_file_prop_xpath = '{DAV:}propstat/d:prop/%s:%s' % (ns, field)
        data = File.build_xml_propfind(fields={ns: [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

    def get_file(self, path):
        """
        Return the File object associated to the path

        :param path: path to the file
        :returns: File object or None
        """
        resp = self.client.with_attr(json_output=False).list_folders(
            path, all_properties=True, depth=0)
        if resp.is_ok:
            if resp.data:
                return resp.data[0]
        return None

    def get_folder(self, path=None):
        """
        Return the File object associated to the path
        If the file (folder or 'collection') doesn't exists, create it.

        :param path: path to the file/folder, if empty use root
        :returns: File object
        """
        fileobj = self.get_file(path)
        if fileobj:
            if not fileobj.isdir():
                raise NextCloudFileConflict(fileobj.href)
        else:
            self.client.create_folder(path)
            fileobj = self.get_file(path)

        return fileobj

    def get_relative_path(self, href):
        """
        Returns relative (to application / user) path

        :param href(str):  file href
        :returns   (str):  relative path
        """
        _app_root = '/'.join([self.API_URL, self.client.user])
        return href[len(_app_root):]


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 ImportError:
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 (
19			timestamp_to_epoch_time,
20			datetime_to_timestamp
21			)
22
23
24			class NextCloudFileConflict(Exception):
25			""" Exception to raise when you try to create a File that alreay exists """
26
27
28			class File(PropertySet):
29			"""
30			Define properties on a WebDav file/folder
31
32			Additionnally, provide an objective CRUD API
33			(that probably consume more energy than fetching specific attributes)
34
35			Example :
36			>>> root = nxc.get_folder() # get root
37			>>> def _list_rec(d, indent=""):
38			>>> # list files recursively
39			>>> print("%s%s%s" % (indent, d.basename(), '/' if d.isdir() else ''))
40			>>> if d.isdir():
41			>>> for i in d.list():
42			>>> _list_rec(i, indent=indent+" ")
43			>>>
44			>>> _list_rec(root)
45			"""
46
47			@staticmethod
48			def _extract_resource_type(file_property):
49			file_type = list(file_property)
50			if file_type:
51			return re.sub('{.*}', '', file_type[0].tag)
52			return None
53
54			_attrs = [
55			Prop('d:getlastmodified'),
56			Prop('d:getetag'),
57			Prop('d:getcontenttype'),
58			Prop('d:resourcetype', parse_xml_value=(lambda p: File._extract_resource_type(p))),
59			Prop('d:getcontentlength'),
60			Prop('oc:id'),
61			Prop('oc:fileid'),
62			Prop('oc:favorite'),
63			Prop('oc:comments-href'),
64			Prop('oc:comments-count'),
65			Prop('oc:comments-unread'),
66			Prop('oc:owner-id'),
67			Prop('oc:owner-display-name'),
68			Prop('oc:share-types'),
69			Prop('oc:checksums'),
70			Prop('oc:size'),
71			Prop('oc:href'),
72			Prop('nc:has-preview')
73			]
74
75			def isfile(self):
76			""" say if the file is a file /!\\ ressourcetype property shall be loaded """
77			return not self.resource_type
78
79			def isdir(self):
80			""" say if the file is a directory /!\\ ressourcetype property shall be loaded """
81			return self.resource_type == self.COLLECTION_RESOURCE_TYPE
82
83			def get_relative_path(self):
84			""" get path relative to user root """
85			return self._wrapper.get_relative_path(self.href)
86
87			def _get_remote_path(self, path=None):
88			_url = self.get_relative_path()
89			return '/'.join([_url, path]) if path else _url
90
91			def basename(self):
92			""" basename """
93			_path = self._get_remote_path()
94			return _path.split('/')[-2] if _path.endswith('/') else _path.split('/')[-1]
95
96			def dirname(self):
97			""" dirname """
98			_path = self._get_remote_path()
99			return '/'.join(_path.split('/')[:-2]) if _path.endswith('/') else '/'.join(_path.split('/')[:-1])
100
101			def __eq__(self, b):
102			return self.href == b.href
103
104			# MINIMAL SET OF CRUD OPERATIONS
105			def get_folder(self, path=None):
106			"""
107			Get folder (see WebDav wrapper)
108			:param path: if empty list current dir
109			:returns: a folder (File object)
110
111			Note : To check if sub folder exists, use get_file method
112			"""
113			return self._wrapper.get_folder(self._get_remote_path(path))
114
115			def get_file(self, path=None):
116			"""
117			Get file (see WebDav wrapper)
118			:param path: if empty list current dir
119			:returns: a file or folder (File object)
120			"""
121			return self._wrapper.get_file(self._get_remote_path(path))
122
123			def list(self, subpath='', filter_rules=None):
124			"""
125			List folder (see WebDav wrapper)
126			:param subpath: if empty list current dir
127			:returns: list of Files
128			"""
129			if filter_rules:
130			resp = self._wrapper.fetch_files_with_filter(
131			path=self._get_remote_path(subpath),
132			filter_rules=filter_rules
133			)
134			else:
135			resp = self._wrapper.list_folders(
136			self._get_remote_path(subpath),
137			depth=1,
138			all_properties=True
139			)
140			if resp.is_ok and resp.data:
141			_dirs = resp.data
142			# remove current dir
143			if _dirs[0] == self:
144			_dirs = _dirs[1:]
145			return _dirs
146			return []
147
148			def upload_file(self, local_filepath, name, timestamp=None):
149			"""
150			Upload file (see WebDav wrapper)
151			:param name: name of the new file
152			:returns: True if success
153			"""
154			resp = self._wrapper.upload_file(local_filepath,
155			self._get_remote_path(name),
156			timestamp=timestamp)
157			return resp.is_ok
158
159			def download(self, name=None, target_dir=None):
160			"""
161			file (see WebDav wrapper)
162			:param name: name of the new file
163			:returns: True if success
164			"""
165			path = self._get_remote_path(name)
166			target_path, _file_info = self._wrapper.download_file(path,
167			target_dir=target_dir)
168			assert os.path.isfile(target_path), "Download failed"
169			return target_path
170
171			def delete(self, subpath=''):
172			"""
173			Delete file or folder (see WebDav wrapper)
174			:param subpath: if empty, delete current file
175			:returns: True if success
176			"""
177			resp = self._wrapper.delete_path(self._get_remote_path(subpath))
178			return resp.is_ok
179
180
181			class WebDAV(WebDAVApiWrapper):
182			""" WebDav API wrapper """
183			API_URL = "/remote.php/dav/files"
184
185			def _get_path(self, path):
186			if path:
187			return '/'.join([self.client.user, path]).replace('//', '/')
188			return self.client.user
189
190			def list_folders(self, path=None, depth=1, all_properties=False,
191			fields=None):
192			"""
193			Get path files list with files properties with given depth
194			(for current user)
195
196			Args:
197			path (str/None): files path
198			depth (int): depth of listing files (directories content for example)
199			all_properties (bool): list all available file properties in Nextcloud
200			fields (str list): file properties to fetch
201
202			Returns:
203			list of dicts if json_output
204			list of File objects if not json_output
205			"""
206			data = File.build_xml_propfind(
207			use_default=all_properties,
208			fields=fields
209			) if (fields or all_properties) else None
210			resp = self.requester.propfind(additional_url=self._get_path(path),
211			headers={'Depth': str(depth)},
212			data=data)
213			return File.from_response(resp, json_output=self.json_output,
214			wrapper=self)
215
216			def download_file(self, path, target_dir=None):
217			"""
218			Download file by path (for current user)
219			File will be saved to working directory
220			path argument must be valid file path
221			Modified time of saved file will be synced with the file properties in Nextcloud
222
223			Exception will be raised if:
224			* path doesn't exist,
225			* path is a directory, or if
226			* file with same name already exists in working directory
227
228			Args:
229			path (str): file path
230
231			Returns:
232			a tuple (target_path, File object)
233			"""
234			if not target_dir:
235			target_dir = './'
236			filename = path.split('/')[(-1)] if '/' in path else path
237			file_data = self.get_file(path)
238			if not file_data:
239			raise ValueError("Given path doesn't exist")
240			file_resource_type = file_data.resource_type
241			if file_resource_type == File.COLLECTION_RESOURCE_TYPE:
242			raise ValueError("This is a collection, please specify file path")
243			if filename in os.listdir(target_dir):
244			raise ValueError(
245			"File with such name already exists in this directory")
246			filename = os.path.join(target_dir, filename)
247			res = self.requester.download(self._get_path(path))
248			with open(filename, 'wb') as f:
249			f.write(res.data)
250
251			# get timestamp of downloaded file from file property on Nextcloud
252			# If it succeeded, set the timestamp to saved local file
253			# If the timestamp string is invalid or broken, the timestamp is downloaded time.
254			file_timestamp_str = file_data.last_modified
255			file_timestamp = timestamp_to_epoch_time(file_timestamp_str)
256			if isinstance(file_timestamp, int):
257			os.utime(filename, (
258			datetime_to_timestamp(datetime.now()),
259			file_timestamp))
260			return (filename, file_data)
261
262			def upload_file(self, local_filepath, remote_filepath, timestamp=None):
263			"""
264			Upload file to Nextcloud storage
265
266			Args:
267			local_filepath (str): path to file on local storage
268			remote_filepath (str): path where to upload file on Nextcloud storage
269			timestamp (int): timestamp of upload file. If None, get time by local file.
270
271			Returns:
272			requester response
273			"""
274			with open(local_filepath, 'rb') as f:
275			file_contents = f.read()
276			if timestamp is None:
277			timestamp = int(os.path.getmtime(local_filepath))
278			return self.upload_file_contents(file_contents, remote_filepath, timestamp)
279
280			def upload_file_contents(self, file_contents, remote_filepath, timestamp=None):
281			"""
282			Upload file to Nextcloud storage
283
284			Args:
285			file_contents (bytes): Bytes the file to be uploaded consists of
286			remote_filepath (str): path where to upload file on Nextcloud storage
287			timestamp (int): mtime of upload file
288
289			Returns:
290			requester response
291			"""
292			return self.requester.put_with_timestamp((self._get_path(remote_filepath)), data=file_contents,
293			timestamp=timestamp)
294
295			def create_folder(self, folder_path):
296			"""
297			Create folder on Nextcloud storage
298
299			Args:
300			folder_path (str): folder path
301
302			Returns:
303			requester response
304			"""
305			return self.requester.make_collection(additional_url=(self._get_path(folder_path)))
306
307			def assure_folder_exists(self, folder_path):
308			"""
309			Create folder on Nextcloud storage, don't do anything if the folder already exists.
310			Args:
311			folder_path (str): folder path
312			Returns:
313			requester response
314			"""
315			self.create_folder(folder_path)
316			return True
317
318			def assure_tree_exists(self, tree_path):
319			"""
320			Make sure that the folder structure on Nextcloud storage exists
321			Args:
322			folder_path (str): The folder tree
323			Returns:
324			requester response
325			"""
326			tree = pathlib.PurePath(tree_path)
327			parents = list(tree.parents)
328			ret = True
329			subfolders = parents[:-1][::-1] + [tree]
330			for subf in subfolders:
331			ret = self.assure_folder_exists(str(subf))
332
333			return ret
334
335			def delete_path(self, path):
336			"""
337			Delete file or folder with all content of given user by path
338
339			Args:
340			path (str): file or folder path to delete
341
342			Returns:
343			requester response
344			"""
345			return self.requester.delete(url=self._get_path(path))
346
347			def move_path(self, path, destination_path, overwrite=False):
348			"""
349			Move file or folder to destination
350
351			Args:
352			path (str): file or folder path to move
353			destionation_path (str): destination where to move
354			overwrite (bool): allow destination path overriding
355
356			Returns:
357			requester response
358			"""
359			return self.requester.move(url=self._get_path(path),
360			destination=self._get_path(
361			destination_path),
362			overwrite=overwrite)
363
364			def copy_path(self, path, destination_path, overwrite=False):
365			"""
366			Copy file or folder to destination
367
368			Args:
369			path (str): file or folder path to copy
370			destionation_path (str): destination where to copy
371			overwrite (bool): allow destination path overriding
372
373			Returns:
374			requester response
375			"""
376			return self.requester.copy(url=self._get_path(path),
377			destination=self._get_path(
378			destination_path),
379			overwrite=overwrite)
380
381			def set_file_property(self, path, update_rules):
382			"""
383			Set file property
384
385			Args:
386			path (str): file or folder path to make favorite
387			update_rules : a dict { namespace: {key : value } }
388
389			Returns:
390			requester response with list<File> in data
391
392			Note :
393			check keys in nextcloud.common.properties.NAMESPACES_MAP for namespace codes
394			check object property xml_name for property name
395			"""
396			data = File.build_xml_propupdate(update_rules)
397			return self.requester.proppatch(additional_url=self._get_path(path), data=data)
398
399			def fetch_files_with_filter(self, path='', filter_rules=''):
400			"""
401			List files according to a filter
402
403			Args:
404			path (str): file or folder path to search
405			filter_rules : a dict { namespace: {key : value } }
406
407			Returns:
408			requester response with list<File> in data
409
410			Note :
411			check keys in nextcloud.common.properties.NAMESPACES_MAP for namespace codes
412			check object property xml_name for property name
413			"""
414			data = File.build_xml_propfind(
415			instr='oc:filter-files', filter_rules=filter_rules)
416			resp = self.requester.report(
417			additional_url=self._get_path(path), data=data)
418			return File.from_response(resp, json_output=self.json_output,
419			wrapper=self)
420
421			def set_favorites(self, path):
422			"""
423			Set files of a user favorite
424
425			Args:
426			path (str): file or folder path to make favorite
427
428			Returns:
429			requester response
430			"""
431			return self.set_file_property(path, {'oc': {'favorite': 1}})
432
433			def list_favorites(self, path=''):
434			"""
435			List favorites (files) of the user
436
437			Args:
438			path (str): file or folder path to search favorite
439
440			Returns:
441			requester response with list<File> in data
442			"""
443			return self.fetch_files_with_filter(path, {'oc': {'favorite': 1}})
444
445			def get_file_property(self, path, field, ns='oc'):
446			"""
447			Fetch asked properties from a file path.
448
449			Args:
450			path (str): file or folder path to make favorite
451			field (str): field name
452
453			Returns:
454			requester response with asked value in data
455			"""
456			if ':' in field:
457			ns, field = field.split(':')
458			get_file_prop_xpath = '{DAV:}propstat/d:prop/%s:%s' % (ns, field)
459			data = File.build_xml_propfind(fields={ns: [field]})
460			resp = self.requester.propfind(additional_url=(self._get_path(path)), headers={'Depth': str(0)},
461			data=data)
462			response_data = resp.data
463			resp.data = None
464			if not resp.is_ok:
465			return resp
466
467			response_xml_data = ET.fromstring(response_data)
468			for xml_data in response_xml_data:
469			for prop in xml_data.findall(get_file_prop_xpath,
470			NAMESPACES_MAP):
471			resp.data = prop.text
472			break
473
474			return resp
475
476			def get_file(self, path):
477			"""
478			Return the File object associated to the path
479
480			:param path: path to the file
481			:returns: File object or None
482			"""
483			resp = self.client.with_attr(json_output=False).list_folders(
484			path, all_properties=True, depth=0)
485			if resp.is_ok:
486			if resp.data:
487			return resp.data[0]
488			return None
489
490			def get_folder(self, path=None):
491			"""
492			Return the File object associated to the path
493			If the file (folder or 'collection') doesn't exists, create it.
494
495			:param path: path to the file/folder, if empty use root
496			:returns: File object
497			"""
498			fileobj = self.get_file(path)
499			if fileobj:
500			if not fileobj.isdir():
501			raise NextCloudFileConflict(fileobj.href)
502			else:
503			self.client.create_folder(path)
504			fileobj = self.get_file(path)
505
506			return fileobj
507
508			def get_relative_path(self, href):
509			"""
510			Returns relative (to application / user) path
511
512			:param href(str): file href
513			:returns (str): relative path
514			"""
515			_app_root = '/'.join([self.API_URL, self.client.user])
516			return href[len(_app_root):]
517

EnterpriseyIntranet / nextcloud-API

Pull Request — master (#69)

WebDAV.upload_file_contents() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like