nextcloud.api_wrappers.webdav.WebDAV.copy_path() - 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 21:16 UTC

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

↳ Parent: nextcloud.api_wrappers.webdav

Complexity

Conditions

Size

Total Lines	16
Code Lines	5

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
eloc	5
dl	0
loc	16
rs	10
c	0
b	0
f	0
cc	1
nop	4

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

EnterpriseyIntranet / nextcloud-API

Pull Request — master (#69)

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

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like