Passed
Pull Request — master (#69)
by
unknown
01:22
created

nextcloud.api_wrappers.webdav.File.delete()   A

Complexity

Conditions 1

Size

Total Lines 8
Code Lines 3

Duplication

Lines 0
Ratio 0 %

Importance

Changes 0
Metric Value
eloc 3
dl 0
loc 8
rs 10
c 0
b 0
f 0
cc 1
nop 2
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=File._extract_resource_type),
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=''):
124
        """
125
        List folder (see WebDav wrapper)
126
        :param subpath: if empty list current dir
127
        :returns: list of Files
128
        """
129
        resp = self._wrapper.list_folders(
130
            self._get_remote_path(subpath),
131
            depth=1,
132
            all_properties=True
133
        )
134
        if resp.is_ok and resp.data:
135
            _dirs = resp.data
136
            # remove current dir
137
            if _dirs[0] == self:
138
                _dirs = _dirs[1:]
139
            return _dirs
140
        return []
141
142
    def upload_file(self, local_filepath, name, timestamp=None):
143
        """
144
        Upload file (see WebDav wrapper)
145
        :param name: name of the new file
146
        :returns: True if success
147
        """
148
        resp = self._wrapper.upload_file(local_filepath,
149
                                         self._get_remote_path(name),
150
                                         timestamp=timestamp)
151
        return resp.is_ok
152
153
    def download(self, name=None, target_dir=None):
154
        """
155
         file (see WebDav wrapper)
156
        :param name: name of the new file
157
        :returns: True if success
158
        """
159
        path = self._get_remote_path(name)
160
        target_path, _file_info = self._wrapper.download_file(path,
161
                                                              target_dir=target_dir)
162
        assert os.path.isfile(target_path), "Download failed"
163
        return target_path
164
165
    def delete(self, subpath=''):
166
        """
167
        Delete file or folder (see WebDav wrapper)
168
        :param subpath: if empty, delete current file
169
        :returns: True if success
170
        """
171
        resp = self._wrapper.delete_path(self._get_remote_path(subpath))
172
        return resp.is_ok
173
174
175
class WebDAV(WebDAVApiWrapper):
176
    """ WebDav API wrapper """
177
    API_URL = "/remote.php/dav/files"
178
179
    def _get_path(self, path):
180
        if path:
181
            return '/'.join([self.client.user, path]).replace('//', '/')
182
        return self.client.user
183
184
    def list_folders(self, path=None, depth=1, all_properties=False,
185
                     fields=None):
186
        """
187
        Get path files list with files properties with given depth
188
        (for current user)
189
190
        Args:
191
            path (str/None): files path
192
            depth (int): depth of listing files (directories content for example)
193
            all_properties (bool): list all available file properties in Nextcloud
194
            fields (str list): file properties to fetch
195
196
        Returns:
197
            list of dicts if json_output
198
            list of File objects if not json_output
199
        """
200
        data = File.build_xml_propfind(
201
            use_default=all_properties,
202
            fields=fields
203
        ) if (fields or all_properties) else None
204
        resp = self.requester.propfind(additional_url=self._get_path(path),
205
                                       headers={'Depth': str(depth)},
206
                                       data=data)
207
        return File.from_response(resp, json_output=self.json_output,
208
                                  wrapper=self)
209
210
    def download_file(self, path, target_dir=None):
211
        """
212
        Download file by path (for current user)
213
        File will be saved to working directory
214
        path argument must be valid file path
215
        Modified time of saved file will be synced with the file properties in Nextcloud
216
217
        Exception will be raised if:
218
            * path doesn't exist,
219
            * path is a directory, or if
220
            * file with same name already exists in working directory
221
222
        Args:
223
            path (str): file path
224
225
        Returns:
226
            a tuple (target_path, File object)
227
        """
228
        if not target_dir:
229
            target_dir = './'
230
        filename = path.split('/')[(-1)] if '/' in path else path
231
        file_data = self.get_file(path)
232
        if not file_data:
233
            raise ValueError("Given path doesn't exist")
234
        file_resource_type = file_data.resource_type
235
        if file_resource_type == File.COLLECTION_RESOURCE_TYPE:
236
            raise ValueError("This is a collection, please specify file path")
237
        if filename in os.listdir(target_dir):
238
            raise ValueError(
239
                "File with such name already exists in this directory")
240
        filename = os.path.join(target_dir, filename)
241
        res = self.requester.download(self._get_path(path))
242
        with open(filename, 'wb') as f:
243
            f.write(res.data)
244
245
        # get timestamp of downloaded file from file property on Nextcloud
246
        # If it succeeded, set the timestamp to saved local file
247
        # If the timestamp string is invalid or broken, the timestamp is downloaded time.
248
        file_timestamp_str = file_data.last_modified
249
        file_timestamp = timestamp_to_epoch_time(file_timestamp_str)
250
        if isinstance(file_timestamp, int):
251
            os.utime(filename, (
252
                datetime_to_timestamp(datetime.now()),
253
                file_timestamp))
254
        return (filename, file_data)
255
256
    def upload_file(self, local_filepath, remote_filepath, timestamp=None):
257
        """
258
        Upload file to Nextcloud storage
259
260
        Args:
261
            local_filepath (str): path to file on local storage
262
            remote_filepath (str): path where to upload file on Nextcloud storage
263
            timestamp (int): timestamp of upload file. If None, get time by local file.
264
265
        Returns:
266
            requester response
267
        """
268
        with open(local_filepath, 'rb') as f:
269
            file_contents = f.read()
270
        if timestamp is None:
271
            timestamp = int(os.path.getmtime(local_filepath))
272
        return self.upload_file_contents(file_contents, remote_filepath, timestamp)
273
274
    def upload_file_contents(self, file_contents, remote_filepath, timestamp=None):
275
        """
276
        Upload file to Nextcloud storage
277
278
        Args:
279
            file_contents (bytes): Bytes the file to be uploaded consists of
280
            remote_filepath (str): path where to upload file on Nextcloud storage
281
            timestamp (int):  mtime of upload file
282
283
        Returns:
284
            requester response
285
        """
286
        return self.requester.put_with_timestamp((self._get_path(remote_filepath)), data=file_contents,
287
                                                 timestamp=timestamp)
288
289
    def create_folder(self, folder_path):
290
        """
291
        Create folder on Nextcloud storage
292
293
        Args:
294
            folder_path (str): folder path
295
296
        Returns:
297
            requester response
298
        """
299
        return self.requester.make_collection(additional_url=(self._get_path(folder_path)))
300
301
    def assure_folder_exists(self, folder_path):
302
        """
303
        Create folder on Nextcloud storage, don't do anything if the folder already exists.
304
        Args:
305
            folder_path (str): folder path
306
        Returns:
307
            requester response
308
        """
309
        self.create_folder(folder_path)
310
        return True
311
312
    def assure_tree_exists(self, tree_path):
313
        """
314
        Make sure that the folder structure on Nextcloud storage exists
315
        Args:
316
            folder_path (str): The folder tree
317
        Returns:
318
            requester response
319
        """
320
        tree = pathlib.PurePath(tree_path)
321
        parents = list(tree.parents)
322
        ret = True
323
        subfolders = parents[:-1][::-1] + [tree]
324
        for subf in subfolders:
325
            ret = self.assure_folder_exists(str(subf))
326
327
        return ret
328
329
    def delete_path(self, path):
330
        """
331
        Delete file or folder with all content of given user by path
332
333
        Args:
334
            path (str): file or folder path to delete
335
336
        Returns:
337
            requester response
338
        """
339
        return self.requester.delete(url=self._get_path(path))
340
341
    def move_path(self, path, destination_path, overwrite=False):
342
        """
343
        Move file or folder to destination
344
345
        Args:
346
            path (str): file or folder path to move
347
            destionation_path (str): destination where to move
348
            overwrite (bool): allow destination path overriding
349
350
        Returns:
351
            requester response
352
        """
353
        return self.requester.move(url=self._get_path(path),
354
                                   destination=self._get_path(
355
                                       destination_path),
356
                                   overwrite=overwrite)
357
358
    def copy_path(self, path, destination_path, overwrite=False):
359
        """
360
        Copy file or folder to destination
361
362
        Args:
363
            path (str): file or folder path to copy
364
            destionation_path (str): destination where to copy
365
            overwrite (bool): allow destination path overriding
366
367
        Returns:
368
            requester response
369
        """
370
        return self.requester.copy(url=self._get_path(path),
371
                                   destination=self._get_path(
372
                                       destination_path),
373
                                   overwrite=overwrite)
374
375
    def set_file_property(self, path, update_rules):
376
        """
377
        Set file property
378
379
        Args:
380
            path (str): file or folder path to make favorite
381
            update_rules : a dict { namespace: {key : value } }
382
383
        Returns:
384
            requester response with <list>File in data
385
386
        Note :
387
            check keys in nextcloud.common.properties.NAMESPACES_MAP for namespace codes
388
            check object property xml_name for property name
389
        """
390
        data = File.build_xml_propupdate(update_rules)
391
        return self.requester.proppatch(additional_url=self._get_path(path), data=data)
392
393
    def list_files_with_filter(self, path='', filter_rules=''):
394
        """
395
        List files according to a filter
396
397
        Args:
398
            path (str): file or folder path to search
399
            filter_rules : a dict { namespace: {key : value } }
400
401
        Returns:
402
            requester response with <list>File in data
403
404
        Note :
405
            check keys in nextcloud.common.properties.NAMESPACES_MAP for namespace codes
406
            check object property xml_name for property name
407
        """
408
        data = File.build_xml_propfind(
409
            instr='oc:filter-files', filter_rules=filter_rules)
410
        resp = self.requester.report(
411
            additional_url=self._get_path(path), data=data)
412
        return File.from_response(resp, json_output=self.json_output,
413
                                  wrapper=self)
414
415
    def set_favorites(self, path):
416
        """
417
        Set files of a user favorite
418
419
        Args:
420
            path (str): file or folder path to make favorite
421
422
        Returns:
423
            requester response
424
        """
425
        return self.set_file_property(path, {'oc': {'favorite': 1}})
426
427
    def list_favorites(self, path=''):
428
        """
429
        List favorites (files) of the user
430
431
        Args:
432
            path (str): file or folder path to search favorite
433
434
        Returns:
435
            requester response with <list>File in data
436
        """
437
        return self.list_files_with_filter(path, {'oc': {'favorite': 1}})
438
439
    def get_file_property(self, path, field, ns='oc'):
440
        """
441
        Fetch asked properties from a file path.
442
443
        Args:
444
            path (str): file or folder path to make favorite
445
            field (str): field name
446
447
        Returns:
448
            requester response with asked value in data
449
        """
450
        if ':' in field:
451
            ns, field = field.split(':')
452
        get_file_prop_xpath = '{DAV:}propstat/d:prop/%s:%s' % (ns, field)
453
        data = File.build_xml_propfind(fields={ns: [field]})
454
        resp = self.requester.propfind(additional_url=(self._get_path(path)), headers={'Depth': str(0)},
455
                                       data=data)
456
        response_data = resp.data
457
        resp.data = None
458
        if not resp.is_ok:
459
            return resp
460
461
        response_xml_data = ET.fromstring(response_data)
462
        for xml_data in response_xml_data:
463
            for prop in xml_data.findall(get_file_prop_xpath,
464
                                         NAMESPACES_MAP):
465
                resp.data = prop.text
466
            break
467
468
        return resp
469
470
    def get_file(self, path):
471
        """
472
        Return the File object associated to the path
473
474
        :param path: path to the file
475
        :returns: File object or None
476
        """
477
        resp = self.client.with_attr(json_output=False).list_folders(
478
            path, all_properties=True, depth=0)
479
        if resp.is_ok:
480
            if resp.data:
481
                return resp.data[0]
482
        return None
483
484
    def get_folder(self, path=None):
485
        """
486
        Return the File object associated to the path
487
        If the file (folder or 'collection') doesn't exists, create it.
488
489
        :param path: path to the file/folder, if empty use root
490
        :returns: File object
491
        """
492
        fileobj = self.get_file(path)
493
        if fileobj:
494
            if not fileobj.isdir():
495
                raise NextCloudFileConflict(fileobj.href)
496
        else:
497
            self.client.create_folder(path)
498
            fileobj = self.get_file(path)
499
500
        return fileobj
501
502
    def get_relative_path(self, href):
503
        """
504
        Returns relative (to application / user) path
505
506
        :param href(str):  file href
507
        :returns   (str):  relative path
508
        """
509
        _app_root = '/'.join([self.API_URL, self.client.user])
510
        return href[len(_app_root):]
511