Statistics
| Branch: | Tag: | Revision:

root / kamaki / clients / pithos / __init__.py @ 438efab2

History | View | Annotate | Download (49.1 kB)

1
# Copyright 2011-2013 GRNET S.A. All rights reserved.
2
#
3
# Redistribution and use in source and binary forms, with or
4
# without modification, are permitted provided that the following
5
# conditions are met:
6
#
7
#   1. Redistributions of source code must retain the above
8
#      copyright notice, this list of conditions and the following
9
#      disclaimer.
10
#
11
#   2. Redistributions in binary form must reproduce the above
12
#      copyright notice, this list of conditions and the following
13
#      disclaimer in the documentation and/or other materials
14
#      provided with the distribution.
15
#
16
# THIS SOFTWARE IS PROVIDED BY GRNET S.A. ``AS IS'' AND ANY EXPRESS
17
# OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED
18
# WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR
19
# PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL GRNET S.A OR
20
# CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
21
# SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT
22
# LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF
23
# USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED
24
# AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT
25
# LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN
26
# ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE
27
# POSSIBILITY OF SUCH DAMAGE.
28
#
29
# The views and conclusions contained in the software and
30
# documentation are those of the authors and should not be
31
# interpreted as representing official policies, either expressed
32
# or implied, of GRNET S.A.
33

    
34
from threading import enumerate as activethreads
35

    
36
from os import fstat
37
from hashlib import new as newhashlib
38
from time import time
39
from StringIO import StringIO
40

    
41
from binascii import hexlify
42

    
43
from kamaki.clients import SilentEvent, sendlog
44
from kamaki.clients.pithos.rest_api import PithosRestClient
45
from kamaki.clients.storage import ClientError
46
from kamaki.clients.utils import path4url, filter_in, readall
47

    
48

    
49
def _pithos_hash(block, blockhash):
50
    h = newhashlib(blockhash)
51
    h.update(block.rstrip('\x00'))
52
    return h.hexdigest()
53

    
54

    
55
def _range_up(start, end, max_value, a_range):
56
    """
57
    :param start: (int) the window bottom
58

59
    :param end: (int) the window top
60

61
    :param max_value: (int) maximum accepted value
62

63
    :param a_range: (str) a range string in the form X[,X'[,X''[...]]]
64
        where X: x|x-y|-x where x < y and x, y natural numbers
65

66
    :returns: (str) a range string cut-off for the start-end range
67
        an empty response means this window is out of range
68
    """
69
    assert start >= 0, '_range_up called w. start(%s) < 0' % start
70
    assert end >= start, '_range_up called w. end(%s) < start(%s)' % (
71
        end, start)
72
    assert end <= max_value, '_range_up called w. max_value(%s) < end(%s)' % (
73
        max_value, end)
74
    if not a_range:
75
        return '%s-%s' % (start, end)
76
    selected = []
77
    for some_range in a_range.split(','):
78
        v0, sep, v1 = some_range.partition('-')
79
        if v0:
80
            v0 = int(v0)
81
            if sep:
82
                v1 = int(v1)
83
                if v1 < start or v0 > end or v1 < v0:
84
                    continue
85
                v0 = v0 if v0 > start else start
86
                v1 = v1 if v1 < end else end
87
                selected.append('%s-%s' % (v0, v1))
88
            elif v0 < start:
89
                continue
90
            else:
91
                v1 = v0 if v0 <= end else end
92
                selected.append('%s-%s' % (start, v1))
93
        else:
94
            v1 = int(v1)
95
            if max_value - v1 > end:
96
                continue
97
            v0 = (max_value - v1) if max_value - v1 > start else start
98
            selected.append('%s-%s' % (v0, end))
99
    return ','.join(selected)
100

    
101

    
102
class PithosClient(PithosRestClient):
103
    """Synnefo Pithos+ API client"""
104

    
105
    def __init__(self, base_url, token, account=None, container=None):
106
        super(PithosClient, self).__init__(base_url, token, account, container)
107

    
108
    def create_container(
109
            self,
110
            container=None, sizelimit=None, versioning=None, metadata=None,
111
            **kwargs):
112
        """
113
        :param container: (str) if not given, self.container is used instead
114

115
        :param sizelimit: (int) container total size limit in bytes
116

117
        :param versioning: (str) can be auto or whatever supported by server
118

119
        :param metadata: (dict) Custom user-defined metadata of the form
120
            { 'name1': 'value1', 'name2': 'value2', ... }
121

122
        :returns: (dict) response headers
123
        """
124
        cnt_back_up = self.container
125
        try:
126
            self.container = container or cnt_back_up
127
            r = self.container_put(
128
                quota=sizelimit, versioning=versioning, metadata=metadata,
129
                **kwargs)
130
            return r.headers
131
        finally:
132
            self.container = cnt_back_up
133

    
134
    def purge_container(self, container=None):
135
        """Delete an empty container and destroy associated blocks
136
        """
137
        cnt_back_up = self.container
138
        try:
139
            self.container = container or cnt_back_up
140
            r = self.container_delete(until=unicode(time()))
141
        finally:
142
            self.container = cnt_back_up
143
        return r.headers
144

    
145
    def upload_object_unchunked(
146
            self, obj, f,
147
            withHashFile=False,
148
            size=None,
149
            etag=None,
150
            content_encoding=None,
151
            content_disposition=None,
152
            content_type=None,
153
            sharing=None,
154
            public=None):
155
        """
156
        :param obj: (str) remote object path
157

158
        :param f: open file descriptor
159

160
        :param withHashFile: (bool)
161

162
        :param size: (int) size of data to upload
163

164
        :param etag: (str)
165

166
        :param content_encoding: (str)
167

168
        :param content_disposition: (str)
169

170
        :param content_type: (str)
171

172
        :param sharing: {'read':[user and/or grp names],
173
            'write':[usr and/or grp names]}
174

175
        :param public: (bool)
176

177
        :returns: (dict) created object metadata
178
        """
179
        self._assert_container()
180

    
181
        if withHashFile:
182
            data = f.read()
183
            try:
184
                import json
185
                data = json.dumps(json.loads(data))
186
            except ValueError:
187
                raise ClientError('"%s" is not json-formated' % f.name, 1)
188
            except SyntaxError:
189
                msg = '"%s" is not a valid hashmap file' % f.name
190
                raise ClientError(msg, 1)
191
            f = StringIO(data)
192
        else:
193
            data = readall(f, size) if size else f.read()
194
        r = self.object_put(
195
            obj,
196
            data=data,
197
            etag=etag,
198
            content_encoding=content_encoding,
199
            content_disposition=content_disposition,
200
            content_type=content_type,
201
            permissions=sharing,
202
            public=public,
203
            success=201)
204
        return r.headers
205

    
206
    def create_object_by_manifestation(
207
            self, obj,
208
            etag=None,
209
            content_encoding=None,
210
            content_disposition=None,
211
            content_type=None,
212
            sharing=None,
213
            public=None):
214
        """
215
        :param obj: (str) remote object path
216

217
        :param etag: (str)
218

219
        :param content_encoding: (str)
220

221
        :param content_disposition: (str)
222

223
        :param content_type: (str)
224

225
        :param sharing: {'read':[user and/or grp names],
226
            'write':[usr and/or grp names]}
227

228
        :param public: (bool)
229

230
        :returns: (dict) created object metadata
231
        """
232
        self._assert_container()
233
        r = self.object_put(
234
            obj,
235
            content_length=0,
236
            etag=etag,
237
            content_encoding=content_encoding,
238
            content_disposition=content_disposition,
239
            content_type=content_type,
240
            permissions=sharing,
241
            public=public,
242
            manifest='%s/%s' % (self.container, obj))
243
        return r.headers
244

    
245
    # upload_* auxiliary methods
246
    def _put_block_async(self, data, hash):
247
        event = SilentEvent(method=self._put_block, data=data, hash=hash)
248
        event.start()
249
        return event
250

    
251
    def _put_block(self, data, hash):
252
        r = self.container_post(
253
            update=True,
254
            content_type='application/octet-stream',
255
            content_length=len(data),
256
            data=data,
257
            format='json')
258
        assert r.json[0] == hash, 'Local hash does not match server'
259

    
260
    def _get_file_block_info(self, fileobj, size=None, cache=None):
261
        """
262
        :param fileobj: (file descriptor) source
263

264
        :param size: (int) size of data to upload from source
265

266
        :param cache: (dict) if provided, cache container info response to
267
        avoid redundant calls
268
        """
269
        if isinstance(cache, dict):
270
            try:
271
                meta = cache[self.container]
272
            except KeyError:
273
                meta = self.get_container_info()
274
                cache[self.container] = meta
275
        else:
276
            meta = self.get_container_info()
277
        blocksize = int(meta['x-container-block-size'])
278
        blockhash = meta['x-container-block-hash']
279
        size = size if size is not None else fstat(fileobj.fileno()).st_size
280
        nblocks = 1 + (size - 1) // blocksize
281
        return (blocksize, blockhash, size, nblocks)
282

    
283
    def _create_object_or_get_missing_hashes(
284
            self, obj, json,
285
            size=None,
286
            format='json',
287
            hashmap=True,
288
            content_type=None,
289
            if_etag_match=None,
290
            if_etag_not_match=None,
291
            content_encoding=None,
292
            content_disposition=None,
293
            permissions=None,
294
            public=None,
295
            success=(201, 409)):
296
        r = self.object_put(
297
            obj,
298
            format='json',
299
            hashmap=True,
300
            content_type=content_type,
301
            json=json,
302
            if_etag_match=if_etag_match,
303
            if_etag_not_match=if_etag_not_match,
304
            content_encoding=content_encoding,
305
            content_disposition=content_disposition,
306
            permissions=permissions,
307
            public=public,
308
            success=success)
309
        return (None if r.status_code == 201 else r.json), r.headers
310

    
311
    def _calculate_blocks_for_upload(
312
            self, blocksize, blockhash, size, nblocks, hashes, hmap, fileobj,
313
            hash_cb=None):
314
        offset = 0
315
        if hash_cb:
316
            hash_gen = hash_cb(nblocks)
317
            hash_gen.next()
318

    
319
        for i in range(nblocks):
320
            block = readall(fileobj, min(blocksize, size - offset))
321
            bytes = len(block)
322
            hash = _pithos_hash(block, blockhash)
323
            hashes.append(hash)
324
            hmap[hash] = (offset, bytes)
325
            offset += bytes
326
            if hash_cb:
327
                hash_gen.next()
328
        msg = ('Failed to calculate uploaded blocks:'
329
               ' Offset and object size do not match')
330
        assert offset == size, msg
331

    
332
    def _upload_missing_blocks(self, missing, hmap, fileobj, upload_gen=None):
333
        """upload missing blocks asynchronously"""
334

    
335
        self._init_thread_limit()
336

    
337
        flying = []
338
        failures = []
339
        for hash in missing:
340
            offset, bytes = hmap[hash]
341
            fileobj.seek(offset)
342
            data = readall(fileobj, bytes)
343
            r = self._put_block_async(data, hash)
344
            flying.append(r)
345
            unfinished = self._watch_thread_limit(flying)
346
            for thread in set(flying).difference(unfinished):
347
                if thread.exception:
348
                    failures.append(thread)
349
                    if isinstance(
350
                            thread.exception,
351
                            ClientError) and thread.exception.status == 502:
352
                        self.POOLSIZE = self._thread_limit
353
                elif thread.isAlive():
354
                    flying.append(thread)
355
                elif upload_gen:
356
                    try:
357
                        upload_gen.next()
358
                    except:
359
                        pass
360
            flying = unfinished
361

    
362
        for thread in flying:
363
            thread.join()
364
            if thread.exception:
365
                failures.append(thread)
366
            elif upload_gen:
367
                try:
368
                    upload_gen.next()
369
                except:
370
                    pass
371

    
372
        return [failure.kwargs['hash'] for failure in failures]
373

    
374
    def upload_object(
375
            self, obj, f,
376
            size=None,
377
            hash_cb=None,
378
            upload_cb=None,
379
            etag=None,
380
            if_etag_match=None,
381
            if_not_exist=None,
382
            content_encoding=None,
383
            content_disposition=None,
384
            content_type=None,
385
            sharing=None,
386
            public=None,
387
            container_info_cache=None):
388
        """Upload an object using multiple connections (threads)
389

390
        :param obj: (str) remote object path
391

392
        :param f: open file descriptor (rb)
393

394
        :param hash_cb: optional progress.bar object for calculating hashes
395

396
        :param upload_cb: optional progress.bar object for uploading
397

398
        :param etag: (str)
399

400
        :param if_etag_match: (str) Push that value to if-match header at file
401
            creation
402

403
        :param if_not_exist: (bool) If true, the file will be uploaded ONLY if
404
            it does not exist remotely, otherwise the operation will fail.
405
            Involves the case of an object with the same path is created while
406
            the object is being uploaded.
407

408
        :param content_encoding: (str)
409

410
        :param content_disposition: (str)
411

412
        :param content_type: (str)
413

414
        :param sharing: {'read':[user and/or grp names],
415
            'write':[usr and/or grp names]}
416

417
        :param public: (bool)
418

419
        :param container_info_cache: (dict) if given, avoid redundant calls to
420
            server for container info (block size and hash information)
421
        """
422
        self._assert_container()
423

    
424
        block_info = (
425
            blocksize, blockhash, size, nblocks) = self._get_file_block_info(
426
                f, size, container_info_cache)
427
        (hashes, hmap, offset) = ([], {}, 0)
428
        if not content_type:
429
            content_type = 'application/octet-stream'
430

    
431
        self._calculate_blocks_for_upload(
432
            *block_info,
433
            hashes=hashes,
434
            hmap=hmap,
435
            fileobj=f,
436
            hash_cb=hash_cb)
437

    
438
        hashmap = dict(bytes=size, hashes=hashes)
439
        missing, obj_headers = self._create_object_or_get_missing_hashes(
440
            obj, hashmap,
441
            content_type=content_type,
442
            size=size,
443
            if_etag_match=if_etag_match,
444
            if_etag_not_match='*' if if_not_exist else None,
445
            content_encoding=content_encoding,
446
            content_disposition=content_disposition,
447
            permissions=sharing,
448
            public=public)
449

    
450
        if missing is None:
451
            return obj_headers
452

    
453
        if upload_cb:
454
            upload_gen = upload_cb(len(missing))
455
            for i in range(len(missing), len(hashmap['hashes']) + 1):
456
                try:
457
                    upload_gen.next()
458
                except:
459
                    upload_gen = None
460
        else:
461
            upload_gen = None
462

    
463
        retries = 7
464
        try:
465
            while retries:
466
                sendlog.info('%s blocks missing' % len(missing))
467
                num_of_blocks = len(missing)
468
                missing = self._upload_missing_blocks(
469
                    missing,
470
                    hmap,
471
                    f,
472
                    upload_gen)
473
                if missing:
474
                    if num_of_blocks == len(missing):
475
                        retries -= 1
476
                    else:
477
                        num_of_blocks = len(missing)
478
                else:
479
                    break
480
            if missing:
481
                try:
482
                    details = ['%s' % thread.exception for thread in missing]
483
                except Exception:
484
                    details = ['Also, failed to read thread exceptions']
485
                raise ClientError(
486
                    '%s blocks failed to upload' % len(missing),
487
                    details=details)
488
        except KeyboardInterrupt:
489
            sendlog.info('- - - wait for threads to finish')
490
            for thread in activethreads():
491
                thread.join()
492
            raise
493

    
494
        r = self.object_put(
495
            obj,
496
            format='json',
497
            hashmap=True,
498
            content_type=content_type,
499
            content_encoding=content_encoding,
500
            if_etag_match=if_etag_match,
501
            if_etag_not_match='*' if if_not_exist else None,
502
            etag=etag,
503
            json=hashmap,
504
            permissions=sharing,
505
            public=public,
506
            success=201)
507
        return r.headers
508

    
509
    def upload_from_string(
510
            self, obj, input_str,
511
            hash_cb=None,
512
            upload_cb=None,
513
            etag=None,
514
            if_etag_match=None,
515
            if_not_exist=None,
516
            content_encoding=None,
517
            content_disposition=None,
518
            content_type=None,
519
            sharing=None,
520
            public=None,
521
            container_info_cache=None):
522
        """Upload an object using multiple connections (threads)
523

524
        :param obj: (str) remote object path
525

526
        :param input_str: (str) upload content
527

528
        :param hash_cb: optional progress.bar object for calculating hashes
529

530
        :param upload_cb: optional progress.bar object for uploading
531

532
        :param etag: (str)
533

534
        :param if_etag_match: (str) Push that value to if-match header at file
535
            creation
536

537
        :param if_not_exist: (bool) If true, the file will be uploaded ONLY if
538
            it does not exist remotely, otherwise the operation will fail.
539
            Involves the case of an object with the same path is created while
540
            the object is being uploaded.
541

542
        :param content_encoding: (str)
543

544
        :param content_disposition: (str)
545

546
        :param content_type: (str)
547

548
        :param sharing: {'read':[user and/or grp names],
549
            'write':[usr and/or grp names]}
550

551
        :param public: (bool)
552

553
        :param container_info_cache: (dict) if given, avoid redundant calls to
554
            server for container info (block size and hash information)
555
        """
556
        self._assert_container()
557

    
558
        blocksize, blockhash, size, nblocks = self._get_file_block_info(
559
                fileobj=None, size=len(input_str), cache=container_info_cache)
560
        (hashes, hmap, offset) = ([], {}, 0)
561
        if not content_type:
562
            content_type = 'application/octet-stream'
563

    
564
        hashes = []
565
        hmap = {}
566
        for blockid in range(nblocks):
567
            start = blockid * blocksize
568
            block = input_str[start: (start + blocksize)]
569
            hashes.append(_pithos_hash(block, blockhash))
570
            hmap[hashes[blockid]] = (start, block)
571

    
572
        hashmap = dict(bytes=size, hashes=hashes)
573
        missing, obj_headers = self._create_object_or_get_missing_hashes(
574
            obj, hashmap,
575
            content_type=content_type,
576
            size=size,
577
            if_etag_match=if_etag_match,
578
            if_etag_not_match='*' if if_not_exist else None,
579
            content_encoding=content_encoding,
580
            content_disposition=content_disposition,
581
            permissions=sharing,
582
            public=public)
583
        if missing is None:
584
            return obj_headers
585
        num_of_missing = len(missing)
586

    
587
        if upload_cb:
588
            self.progress_bar_gen = upload_cb(nblocks)
589
            for i in range(nblocks + 1 - num_of_missing):
590
                self._cb_next()
591

    
592
        tries = 7
593
        old_failures = 0
594
        try:
595
            while tries and missing:
596
                flying = []
597
                failures = []
598
                for hash in missing:
599
                    offset, block = hmap[hash]
600
                    bird = self._put_block_async(block, hash)
601
                    flying.append(bird)
602
                    unfinished = self._watch_thread_limit(flying)
603
                    for thread in set(flying).difference(unfinished):
604
                        if thread.exception:
605
                            failures.append(thread.kwargs['hash'])
606
                        if thread.isAlive():
607
                            flying.append(thread)
608
                        else:
609
                            self._cb_next()
610
                    flying = unfinished
611
                for thread in flying:
612
                    thread.join()
613
                    if thread.exception:
614
                        failures.append(thread.kwargs['hash'])
615
                    self._cb_next()
616
                missing = failures
617
                if missing and len(missing) == old_failures:
618
                    tries -= 1
619
                old_failures = len(missing)
620
            if missing:
621
                raise ClientError(
622
                    '%s blocks failed to upload' % len(missing),
623
                    details=['%s' % thread.exception for thread in missing])
624
        except KeyboardInterrupt:
625
            sendlog.info('- - - wait for threads to finish')
626
            for thread in activethreads():
627
                thread.join()
628
            raise
629

    
630
        r = self.object_put(
631
            obj,
632
            format='json',
633
            hashmap=True,
634
            content_type=content_type,
635
            content_encoding=content_encoding,
636
            if_etag_match=if_etag_match,
637
            if_etag_not_match='*' if if_not_exist else None,
638
            etag=etag,
639
            json=hashmap,
640
            permissions=sharing,
641
            public=public,
642
            success=201)
643
        return r.headers
644

    
645
    # download_* auxiliary methods
646
    def _get_remote_blocks_info(self, obj, **restargs):
647
        #retrieve object hashmap
648
        myrange = restargs.pop('data_range', None)
649
        hashmap = self.get_object_hashmap(obj, **restargs)
650
        restargs['data_range'] = myrange
651
        blocksize = int(hashmap['block_size'])
652
        blockhash = hashmap['block_hash']
653
        total_size = hashmap['bytes']
654
        #assert total_size/blocksize + 1 == len(hashmap['hashes'])
655
        map_dict = {}
656
        for i, h in enumerate(hashmap['hashes']):
657
            #  map_dict[h] = i   CHAGE
658
            if h in map_dict:
659
                map_dict[h].append(i)
660
            else:
661
                map_dict[h] = [i]
662
        return (blocksize, blockhash, total_size, hashmap['hashes'], map_dict)
663

    
664
    def _dump_blocks_sync(
665
            self, obj, remote_hashes, blocksize, total_size, dst, crange,
666
            **args):
667
        if not total_size:
668
            return
669
        for blockid, blockhash in enumerate(remote_hashes):
670
            if blockhash:
671
                start = blocksize * blockid
672
                is_last = start + blocksize > total_size
673
                end = (total_size - 1) if is_last else (start + blocksize - 1)
674
                data_range = _range_up(start, end, total_size, crange)
675
                if not data_range:
676
                    self._cb_next()
677
                    continue
678
                args['data_range'] = 'bytes=%s' % data_range
679
                r = self.object_get(obj, success=(200, 206), **args)
680
                self._cb_next()
681
                dst.write(r.content)
682
                dst.flush()
683

    
684
    def _get_block_async(self, obj, **args):
685
        event = SilentEvent(self.object_get, obj, success=(200, 206), **args)
686
        event.start()
687
        return event
688

    
689
    def _hash_from_file(self, fp, start, size, blockhash):
690
        fp.seek(start)
691
        block = readall(fp, size)
692
        h = newhashlib(blockhash)
693
        h.update(block.strip('\x00'))
694
        return hexlify(h.digest())
695

    
696
    def _thread2file(self, flying, blockids, local_file, offset=0, **restargs):
697
        """write the results of a greenleted rest call to a file
698

699
        :param offset: the offset of the file up to blocksize
700
        - e.g. if the range is 10-100, all blocks will be written to
701
        normal_position - 10
702
        """
703
        for key, g in flying.items():
704
            if g.isAlive():
705
                continue
706
            if g.exception:
707
                raise g.exception
708
            block = g.value.content
709
            for block_start in blockids[key]:
710
                local_file.seek(block_start + offset)
711
                local_file.write(block)
712
                self._cb_next()
713
            flying.pop(key)
714
            blockids.pop(key)
715
        local_file.flush()
716

    
717
    def _dump_blocks_async(
718
            self, obj, remote_hashes, blocksize, total_size, local_file,
719
            blockhash=None, resume=False, filerange=None, **restargs):
720
        file_size = fstat(local_file.fileno()).st_size if resume else 0
721
        flying = dict()
722
        blockid_dict = dict()
723
        offset = 0
724

    
725
        self._init_thread_limit()
726
        for block_hash, blockids in remote_hashes.items():
727
            blockids = [blk * blocksize for blk in blockids]
728
            unsaved = [blk for blk in blockids if not (
729
                blk < file_size and block_hash == self._hash_from_file(
730
                        local_file, blk, blocksize, blockhash))]
731
            self._cb_next(len(blockids) - len(unsaved))
732
            if unsaved:
733
                key = unsaved[0]
734
                self._watch_thread_limit(flying.values())
735
                self._thread2file(
736
                    flying, blockid_dict, local_file, offset,
737
                    **restargs)
738
                end = total_size - 1 if (
739
                    key + blocksize > total_size) else key + blocksize - 1
740
                if end < key:
741
                    self._cb_next()
742
                    continue
743
                data_range = _range_up(key, end, total_size, filerange)
744
                if not data_range:
745
                    self._cb_next()
746
                    continue
747
                restargs[
748
                    'async_headers'] = {'Range': 'bytes=%s' % data_range}
749
                flying[key] = self._get_block_async(obj, **restargs)
750
                blockid_dict[key] = unsaved
751

    
752
        for thread in flying.values():
753
            thread.join()
754
        self._thread2file(flying, blockid_dict, local_file, offset, **restargs)
755

    
756
    def download_object(
757
            self, obj, dst,
758
            download_cb=None,
759
            version=None,
760
            resume=False,
761
            range_str=None,
762
            if_match=None,
763
            if_none_match=None,
764
            if_modified_since=None,
765
            if_unmodified_since=None):
766
        """Download an object (multiple connections, random blocks)
767

768
        :param obj: (str) remote object path
769

770
        :param dst: open file descriptor (wb+)
771

772
        :param download_cb: optional progress.bar object for downloading
773

774
        :param version: (str) file version
775

776
        :param resume: (bool) if set, preserve already downloaded file parts
777

778
        :param range_str: (str) from, to are file positions (int) in bytes
779

780
        :param if_match: (str)
781

782
        :param if_none_match: (str)
783

784
        :param if_modified_since: (str) formated date
785

786
        :param if_unmodified_since: (str) formated date"""
787
        restargs = dict(
788
            version=version,
789
            data_range=None if range_str is None else 'bytes=%s' % range_str,
790
            if_match=if_match,
791
            if_none_match=if_none_match,
792
            if_modified_since=if_modified_since,
793
            if_unmodified_since=if_unmodified_since)
794

    
795
        (
796
            blocksize,
797
            blockhash,
798
            total_size,
799
            hash_list,
800
            remote_hashes) = self._get_remote_blocks_info(obj, **restargs)
801
        assert total_size >= 0
802

    
803
        if download_cb:
804
            self.progress_bar_gen = download_cb(len(hash_list))
805
            self._cb_next()
806

    
807
        if dst.isatty():
808
            self._dump_blocks_sync(
809
                obj,
810
                hash_list,
811
                blocksize,
812
                total_size,
813
                dst,
814
                range_str,
815
                **restargs)
816
        else:
817
            self._dump_blocks_async(
818
                obj,
819
                remote_hashes,
820
                blocksize,
821
                total_size,
822
                dst,
823
                blockhash,
824
                resume,
825
                range_str,
826
                **restargs)
827
            if not range_str:
828
                dst.truncate(total_size)
829

    
830
        self._complete_cb()
831

    
832
    def download_to_string(
833
            self, obj,
834
            download_cb=None,
835
            version=None,
836
            range_str=None,
837
            if_match=None,
838
            if_none_match=None,
839
            if_modified_since=None,
840
            if_unmodified_since=None):
841
        """Download an object to a string (multiple connections). This method
842
        uses threads for http requests, but stores all content in memory.
843

844
        :param obj: (str) remote object path
845

846
        :param download_cb: optional progress.bar object for downloading
847

848
        :param version: (str) file version
849

850
        :param range_str: (str) from, to are file positions (int) in bytes
851

852
        :param if_match: (str)
853

854
        :param if_none_match: (str)
855

856
        :param if_modified_since: (str) formated date
857

858
        :param if_unmodified_since: (str) formated date
859

860
        :returns: (str) the whole object contents
861
        """
862
        restargs = dict(
863
            version=version,
864
            data_range=None if range_str is None else 'bytes=%s' % range_str,
865
            if_match=if_match,
866
            if_none_match=if_none_match,
867
            if_modified_since=if_modified_since,
868
            if_unmodified_since=if_unmodified_since)
869

    
870
        (
871
            blocksize,
872
            blockhash,
873
            total_size,
874
            hash_list,
875
            remote_hashes) = self._get_remote_blocks_info(obj, **restargs)
876
        assert total_size >= 0
877

    
878
        if download_cb:
879
            self.progress_bar_gen = download_cb(len(hash_list))
880
            self._cb_next()
881

    
882
        num_of_blocks = len(remote_hashes)
883
        ret = [''] * num_of_blocks
884
        self._init_thread_limit()
885
        flying = dict()
886
        try:
887
            for blockid, blockhash in enumerate(remote_hashes):
888
                start = blocksize * blockid
889
                is_last = start + blocksize > total_size
890
                end = (total_size - 1) if is_last else (start + blocksize - 1)
891
                data_range_str = _range_up(start, end, end, range_str)
892
                if data_range_str:
893
                    self._watch_thread_limit(flying.values())
894
                    restargs['data_range'] = 'bytes=%s' % data_range_str
895
                    flying[blockid] = self._get_block_async(obj, **restargs)
896
                for runid, thread in flying.items():
897
                    if (blockid + 1) == num_of_blocks:
898
                        thread.join()
899
                    elif thread.isAlive():
900
                        continue
901
                    if thread.exception:
902
                        raise thread.exception
903
                    ret[runid] = thread.value.content
904
                    self._cb_next()
905
                    flying.pop(runid)
906
            return ''.join(ret)
907
        except KeyboardInterrupt:
908
            sendlog.info('- - - wait for threads to finish')
909
            for thread in activethreads():
910
                thread.join()
911

    
912
    #Command Progress Bar method
913
    def _cb_next(self, step=1):
914
        if hasattr(self, 'progress_bar_gen'):
915
            try:
916
                for i in xrange(step):
917
                    self.progress_bar_gen.next()
918
            except:
919
                pass
920

    
921
    def _complete_cb(self):
922
        while True:
923
            try:
924
                self.progress_bar_gen.next()
925
            except:
926
                break
927

    
928
    def get_object_hashmap(
929
            self, obj,
930
            version=None,
931
            if_match=None,
932
            if_none_match=None,
933
            if_modified_since=None,
934
            if_unmodified_since=None):
935
        """
936
        :param obj: (str) remote object path
937

938
        :param if_match: (str)
939

940
        :param if_none_match: (str)
941

942
        :param if_modified_since: (str) formated date
943

944
        :param if_unmodified_since: (str) formated date
945

946
        :returns: (list)
947
        """
948
        try:
949
            r = self.object_get(
950
                obj,
951
                hashmap=True,
952
                version=version,
953
                if_etag_match=if_match,
954
                if_etag_not_match=if_none_match,
955
                if_modified_since=if_modified_since,
956
                if_unmodified_since=if_unmodified_since)
957
        except ClientError as err:
958
            if err.status == 304 or err.status == 412:
959
                return {}
960
            raise
961
        return r.json
962

    
963
    def set_account_group(self, group, usernames):
964
        """
965
        :param group: (str)
966

967
        :param usernames: (list)
968
        """
969
        r = self.account_post(update=True, groups={group: usernames})
970
        return r
971

    
972
    def del_account_group(self, group):
973
        """
974
        :param group: (str)
975
        """
976
        self.account_post(update=True, groups={group: []})
977

    
978
    def get_account_info(self, until=None):
979
        """
980
        :param until: (str) formated date
981

982
        :returns: (dict)
983
        """
984
        r = self.account_head(until=until)
985
        if r.status_code == 401:
986
            raise ClientError("No authorization", status=401)
987
        return r.headers
988

    
989
    def get_account_quota(self):
990
        """
991
        :returns: (dict)
992
        """
993
        return filter_in(
994
            self.get_account_info(),
995
            'X-Account-Policy-Quota',
996
            exactMatch=True)
997

    
998
    #def get_account_versioning(self):
999
    #    """
1000
    #    :returns: (dict)
1001
    #    """
1002
    #    return filter_in(
1003
    #        self.get_account_info(),
1004
    #        'X-Account-Policy-Versioning',
1005
    #        exactMatch=True)
1006

    
1007
    def get_account_meta(self, until=None):
1008
        """
1009
        :param until: (str) formated date
1010

1011
        :returns: (dict)
1012
        """
1013
        return filter_in(self.get_account_info(until=until), 'X-Account-Meta-')
1014

    
1015
    def get_account_group(self):
1016
        """
1017
        :returns: (dict)
1018
        """
1019
        return filter_in(self.get_account_info(), 'X-Account-Group-')
1020

    
1021
    def set_account_meta(self, metapairs):
1022
        """
1023
        :param metapairs: (dict) {key1:val1, key2:val2, ...}
1024
        """
1025
        assert(type(metapairs) is dict)
1026
        r = self.account_post(update=True, metadata=metapairs)
1027
        return r.headers
1028

    
1029
    def del_account_meta(self, metakey):
1030
        """
1031
        :param metakey: (str) metadatum key
1032
        """
1033
        r = self.account_post(update=True, metadata={metakey: ''})
1034
        return r.headers
1035

    
1036
    #def set_account_quota(self, quota):
1037
    #    """
1038
    #    :param quota: (int)
1039
    #    """
1040
    #    self.account_post(update=True, quota=quota)
1041

    
1042
    #def set_account_versioning(self, versioning):
1043
    #    """
1044
    #    :param versioning: (str)
1045
    #    """
1046
    #    r = self.account_post(update=True, versioning=versioning)
1047
    #    return r.headers
1048

    
1049
    def list_containers(self):
1050
        """
1051
        :returns: (dict)
1052
        """
1053
        r = self.account_get()
1054
        return r.json
1055

    
1056
    def del_container(self, until=None, delimiter=None):
1057
        """
1058
        :param until: (str) formated date
1059

1060
        :param delimiter: (str) with / empty container
1061

1062
        :raises ClientError: 404 Container does not exist
1063

1064
        :raises ClientError: 409 Container is not empty
1065
        """
1066
        self._assert_container()
1067
        r = self.container_delete(
1068
            until=until,
1069
            delimiter=delimiter,
1070
            success=(204, 404, 409))
1071
        if r.status_code == 404:
1072
            raise ClientError(
1073
                'Container "%s" does not exist' % self.container,
1074
                r.status_code)
1075
        elif r.status_code == 409:
1076
            raise ClientError(
1077
                'Container "%s" is not empty' % self.container,
1078
                r.status_code)
1079
        return r.headers
1080

    
1081
    def get_container_versioning(self, container=None):
1082
        """
1083
        :param container: (str)
1084

1085
        :returns: (dict)
1086
        """
1087
        cnt_back_up = self.container
1088
        try:
1089
            self.container = container or cnt_back_up
1090
            return filter_in(
1091
                self.get_container_info(),
1092
                'X-Container-Policy-Versioning')
1093
        finally:
1094
            self.container = cnt_back_up
1095

    
1096
    def get_container_limit(self, container=None):
1097
        """
1098
        :param container: (str)
1099

1100
        :returns: (dict)
1101
        """
1102
        cnt_back_up = self.container
1103
        try:
1104
            self.container = container or cnt_back_up
1105
            return filter_in(
1106
                self.get_container_info(),
1107
                'X-Container-Policy-Quota')
1108
        finally:
1109
            self.container = cnt_back_up
1110

    
1111
    def get_container_info(self, until=None):
1112
        """
1113
        :param until: (str) formated date
1114

1115
        :returns: (dict)
1116

1117
        :raises ClientError: 404 Container not found
1118
        """
1119
        try:
1120
            r = self.container_head(until=until)
1121
        except ClientError as err:
1122
            err.details.append('for container %s' % self.container)
1123
            raise err
1124
        return r.headers
1125

    
1126
    def get_container_meta(self, until=None):
1127
        """
1128
        :param until: (str) formated date
1129

1130
        :returns: (dict)
1131
        """
1132
        return filter_in(
1133
            self.get_container_info(until=until),
1134
            'X-Container-Meta')
1135

    
1136
    def get_container_object_meta(self, until=None):
1137
        """
1138
        :param until: (str) formated date
1139

1140
        :returns: (dict)
1141
        """
1142
        return filter_in(
1143
            self.get_container_info(until=until),
1144
            'X-Container-Object-Meta')
1145

    
1146
    def set_container_meta(self, metapairs):
1147
        """
1148
        :param metapairs: (dict) {key1:val1, key2:val2, ...}
1149
        """
1150
        assert(type(metapairs) is dict)
1151
        r = self.container_post(update=True, metadata=metapairs)
1152
        return r.headers
1153

    
1154
    def del_container_meta(self, metakey):
1155
        """
1156
        :param metakey: (str) metadatum key
1157

1158
        :returns: (dict) response headers
1159
        """
1160
        r = self.container_post(update=True, metadata={metakey: ''})
1161
        return r.headers
1162

    
1163
    def set_container_limit(self, limit):
1164
        """
1165
        :param limit: (int)
1166
        """
1167
        r = self.container_post(update=True, quota=limit)
1168
        return r.headers
1169

    
1170
    def set_container_versioning(self, versioning):
1171
        """
1172
        :param versioning: (str)
1173
        """
1174
        r = self.container_post(update=True, versioning=versioning)
1175
        return r.headers
1176

    
1177
    def del_object(self, obj, until=None, delimiter=None):
1178
        """
1179
        :param obj: (str) remote object path
1180

1181
        :param until: (str) formated date
1182

1183
        :param delimiter: (str)
1184
        """
1185
        self._assert_container()
1186
        r = self.object_delete(obj, until=until, delimiter=delimiter)
1187
        return r.headers
1188

    
1189
    def set_object_meta(self, obj, metapairs):
1190
        """
1191
        :param obj: (str) remote object path
1192

1193
        :param metapairs: (dict) {key1:val1, key2:val2, ...}
1194
        """
1195
        assert(type(metapairs) is dict)
1196
        r = self.object_post(obj, update=True, metadata=metapairs)
1197
        return r.headers
1198

    
1199
    def del_object_meta(self, obj, metakey):
1200
        """
1201
        :param obj: (str) remote object path
1202

1203
        :param metakey: (str) metadatum key
1204
        """
1205
        r = self.object_post(obj, update=True, metadata={metakey: ''})
1206
        return r.headers
1207

    
1208
    def publish_object(self, obj):
1209
        """
1210
        :param obj: (str) remote object path
1211

1212
        :returns: (str) access url
1213
        """
1214
        self.object_post(obj, update=True, public=True)
1215
        info = self.get_object_info(obj)
1216
        return info['x-object-public']
1217
        pref, sep, rest = self.base_url.partition('//')
1218
        base = rest.split('/')[0]
1219
        return '%s%s%s/%s' % (pref, sep, base, info['x-object-public'])
1220

    
1221
    def unpublish_object(self, obj):
1222
        """
1223
        :param obj: (str) remote object path
1224
        """
1225
        r = self.object_post(obj, update=True, public=False)
1226
        return r.headers
1227

    
1228
    def get_object_info(self, obj, version=None):
1229
        """
1230
        :param obj: (str) remote object path
1231

1232
        :param version: (str)
1233

1234
        :returns: (dict)
1235
        """
1236
        try:
1237
            r = self.object_head(obj, version=version)
1238
            return r.headers
1239
        except ClientError as ce:
1240
            if ce.status == 404:
1241
                raise ClientError('Object %s not found' % obj, status=404)
1242
            raise
1243

    
1244
    def get_object_meta(self, obj, version=None):
1245
        """
1246
        :param obj: (str) remote object path
1247

1248
        :param version: (str)
1249

1250
        :returns: (dict)
1251
        """
1252
        return filter_in(
1253
            self.get_object_info(obj, version=version),
1254
            'X-Object-Meta')
1255

    
1256
    def get_object_sharing(self, obj):
1257
        """
1258
        :param obj: (str) remote object path
1259

1260
        :returns: (dict)
1261
        """
1262
        r = filter_in(
1263
            self.get_object_info(obj),
1264
            'X-Object-Sharing',
1265
            exactMatch=True)
1266
        reply = {}
1267
        if len(r) > 0:
1268
            perms = r['x-object-sharing'].split(';')
1269
            for perm in perms:
1270
                try:
1271
                    perm.index('=')
1272
                except ValueError:
1273
                    raise ClientError('Incorrect reply format')
1274
                (key, val) = perm.strip().split('=')
1275
                reply[key] = val
1276
        return reply
1277

    
1278
    def set_object_sharing(
1279
            self, obj,
1280
            read_permission=False, write_permission=False):
1281
        """Give read/write permisions to an object.
1282

1283
        :param obj: (str) remote object path
1284

1285
        :param read_permission: (list - bool) users and user groups that get
1286
            read permission for this object - False means all previous read
1287
            permissions will be removed
1288

1289
        :param write_permission: (list - bool) of users and user groups to get
1290
           write permission for this object - False means all previous write
1291
           permissions will be removed
1292

1293
        :returns: (dict) response headers
1294
        """
1295

    
1296
        perms = dict(read=read_permission or '', write=write_permission or '')
1297
        r = self.object_post(obj, update=True, permissions=perms)
1298
        return r.headers
1299

    
1300
    def del_object_sharing(self, obj):
1301
        """
1302
        :param obj: (str) remote object path
1303
        """
1304
        return self.set_object_sharing(obj)
1305

    
1306
    def append_object(self, obj, source_file, upload_cb=None):
1307
        """
1308
        :param obj: (str) remote object path
1309

1310
        :param source_file: open file descriptor
1311

1312
        :param upload_db: progress.bar for uploading
1313
        """
1314
        self._assert_container()
1315
        meta = self.get_container_info()
1316
        blocksize = int(meta['x-container-block-size'])
1317
        filesize = fstat(source_file.fileno()).st_size
1318
        nblocks = 1 + (filesize - 1) // blocksize
1319
        offset = 0
1320
        headers = {}
1321
        if upload_cb:
1322
            self.progress_bar_gen = upload_cb(nblocks)
1323
            self._cb_next()
1324
        flying = {}
1325
        self._init_thread_limit()
1326
        try:
1327
            for i in range(nblocks):
1328
                block = source_file.read(min(blocksize, filesize - offset))
1329
                offset += len(block)
1330

    
1331
                self._watch_thread_limit(flying.values())
1332
                unfinished = {}
1333
                flying[i] = SilentEvent(
1334
                    method=self.object_post,
1335
                    obj=obj,
1336
                    update=True,
1337
                    content_range='bytes */*',
1338
                    content_type='application/octet-stream',
1339
                    content_length=len(block),
1340
                    data=block)
1341
                flying[i].start()
1342

    
1343
                for key, thread in flying.items():
1344
                    if thread.isAlive():
1345
                        if i < nblocks:
1346
                            unfinished[key] = thread
1347
                            continue
1348
                        thread.join()
1349
                    if thread.exception:
1350
                        raise thread.exception
1351
                    headers[key] = thread.value.headers
1352
                    self._cb_next()
1353
                flying = unfinished
1354
        except KeyboardInterrupt:
1355
            sendlog.info('- - - wait for threads to finish')
1356
            for thread in activethreads():
1357
                thread.join()
1358
        finally:
1359
            from time import sleep
1360
            sleep(2 * len(activethreads()))
1361
        return headers.values()
1362

    
1363
    def truncate_object(self, obj, upto_bytes):
1364
        """
1365
        :param obj: (str) remote object path
1366

1367
        :param upto_bytes: max number of bytes to leave on file
1368

1369
        :returns: (dict) response headers
1370
        """
1371
        r = self.object_post(
1372
            obj,
1373
            update=True,
1374
            content_range='bytes 0-%s/*' % upto_bytes,
1375
            content_type='application/octet-stream',
1376
            object_bytes=upto_bytes,
1377
            source_object=path4url(self.container, obj))
1378
        return r.headers
1379

    
1380
    def overwrite_object(self, obj, start, end, source_file, upload_cb=None):
1381
        """Overwrite a part of an object from local source file
1382

1383
        :param obj: (str) remote object path
1384

1385
        :param start: (int) position in bytes to start overwriting from
1386

1387
        :param end: (int) position in bytes to stop overwriting at
1388

1389
        :param source_file: open file descriptor
1390

1391
        :param upload_db: progress.bar for uploading
1392
        """
1393

    
1394
        r = self.get_object_info(obj)
1395
        rf_size = int(r['content-length'])
1396
        if rf_size < int(start):
1397
            raise ClientError(
1398
                'Range start exceeds file size',
1399
                status=416)
1400
        elif rf_size < int(end):
1401
            raise ClientError(
1402
                'Range end exceeds file size',
1403
                status=416)
1404
        self._assert_container()
1405
        meta = self.get_container_info()
1406
        blocksize = int(meta['x-container-block-size'])
1407
        filesize = fstat(source_file.fileno()).st_size
1408
        datasize = int(end) - int(start) + 1
1409
        nblocks = 1 + (datasize - 1) // blocksize
1410
        offset = 0
1411
        if upload_cb:
1412
            self.progress_bar_gen = upload_cb(nblocks)
1413
            self._cb_next()
1414
        headers = []
1415
        for i in range(nblocks):
1416
            read_size = min(blocksize, filesize - offset, datasize - offset)
1417
            block = source_file.read(read_size)
1418
            r = self.object_post(
1419
                obj,
1420
                update=True,
1421
                content_type='application/octet-stream',
1422
                content_length=len(block),
1423
                content_range='bytes %s-%s/*' % (
1424
                    start + offset,
1425
                    start + offset + len(block) - 1),
1426
                data=block)
1427
            headers.append(dict(r.headers))
1428
            offset += len(block)
1429

    
1430
            self._cb_next
1431
        return headers
1432

    
1433
    def copy_object(
1434
            self, src_container, src_object, dst_container,
1435
            dst_object=None,
1436
            source_version=None,
1437
            source_account=None,
1438
            public=False,
1439
            content_type=None,
1440
            delimiter=None):
1441
        """
1442
        :param src_container: (str) source container
1443

1444
        :param src_object: (str) source object path
1445

1446
        :param dst_container: (str) destination container
1447

1448
        :param dst_object: (str) destination object path
1449

1450
        :param source_version: (str) source object version
1451

1452
        :param source_account: (str) account to copy from
1453

1454
        :param public: (bool)
1455

1456
        :param content_type: (str)
1457

1458
        :param delimiter: (str)
1459

1460
        :returns: (dict) response headers
1461
        """
1462
        self._assert_account()
1463
        self.container = dst_container
1464
        src_path = path4url(src_container, src_object)
1465
        r = self.object_put(
1466
            dst_object or src_object,
1467
            success=201,
1468
            copy_from=src_path,
1469
            content_length=0,
1470
            source_version=source_version,
1471
            source_account=source_account,
1472
            public=public,
1473
            content_type=content_type,
1474
            delimiter=delimiter)
1475
        return r.headers
1476

    
1477
    def move_object(
1478
            self, src_container, src_object, dst_container,
1479
            dst_object=False,
1480
            source_account=None,
1481
            source_version=None,
1482
            public=False,
1483
            content_type=None,
1484
            delimiter=None):
1485
        """
1486
        :param src_container: (str) source container
1487

1488
        :param src_object: (str) source object path
1489

1490
        :param dst_container: (str) destination container
1491

1492
        :param dst_object: (str) destination object path
1493

1494
        :param source_account: (str) account to move from
1495

1496
        :param source_version: (str) source object version
1497

1498
        :param public: (bool)
1499

1500
        :param content_type: (str)
1501

1502
        :param delimiter: (str)
1503

1504
        :returns: (dict) response headers
1505
        """
1506
        self._assert_account()
1507
        self.container = dst_container
1508
        dst_object = dst_object or src_object
1509
        src_path = path4url(src_container, src_object)
1510
        r = self.object_put(
1511
            dst_object,
1512
            success=201,
1513
            move_from=src_path,
1514
            content_length=0,
1515
            source_account=source_account,
1516
            source_version=source_version,
1517
            public=public,
1518
            content_type=content_type,
1519
            delimiter=delimiter)
1520
        return r.headers
1521

    
1522
    def get_sharing_accounts(self, limit=None, marker=None, *args, **kwargs):
1523
        """Get accounts that share with self.account
1524

1525
        :param limit: (str)
1526

1527
        :param marker: (str)
1528

1529
        :returns: (dict)
1530
        """
1531
        self._assert_account()
1532

    
1533
        self.set_param('format', 'json')
1534
        self.set_param('limit', limit, iff=limit is not None)
1535
        self.set_param('marker', marker, iff=marker is not None)
1536

    
1537
        path = ''
1538
        success = kwargs.pop('success', (200, 204))
1539
        r = self.get(path, *args, success=success, **kwargs)
1540
        return r.json
1541

    
1542
    def get_object_versionlist(self, obj):
1543
        """
1544
        :param obj: (str) remote object path
1545

1546
        :returns: (list)
1547
        """
1548
        self._assert_container()
1549
        r = self.object_get(obj, format='json', version='list')
1550
        return r.json['versions']