arvados.collection

   1# Copyright (C) The Arvados Authors. All rights reserved.
   2#
   3# SPDX-License-Identifier: Apache-2.0
   4"""Tools to work with Arvados collections
   5
   6This module provides high-level interfaces to create, read, and update
   7Arvados collections. Most users will want to instantiate `Collection`
   8objects, and use methods like `Collection.open` and `Collection.mkdirs` to
   9read and write data in the collection. Refer to the Arvados Python SDK
  10cookbook for [an introduction to using the Collection class][cookbook].
  11
  12[cookbook]: https://doc.arvados.org/sdk/python/cookbook.html#working-with-collections
  13"""
  14
  15import ciso8601
  16import datetime
  17import errno
  18import functools
  19import hashlib
  20import io
  21import logging
  22import os
  23import re
  24import sys
  25import threading
  26import time
  27
  28from collections import deque
  29from stat import *
  30
  31from ._internal import streams
  32from .api import ThreadSafeAPIClient
  33from .arvfile import split, _FileLikeObjectBase, ArvadosFile, ArvadosFileWriter, ArvadosFileReader, WrappableFile, _BlockManager, synchronized, must_be_writable, NoopLock, ADD, DEL, MOD, TOK, WRITE
  34from .keep import KeepLocator, KeepClient
  35import arvados.config as config
  36import arvados.errors as errors
  37import arvados.util
  38import arvados.events as events
  39from arvados.retry import retry_method
  40
  41from typing import (
  42    Any,
  43    Callable,
  44    Dict,
  45    IO,
  46    Iterator,
  47    List,
  48    Mapping,
  49    Optional,
  50    Tuple,
  51    Union,
  52)
  53
  54if sys.version_info < (3, 8):
  55    from typing_extensions import Literal
  56else:
  57    from typing import Literal
  58
  59_logger = logging.getLogger('arvados.collection')
  60
  61
  62FILE = "file"
  63"""`create_type` value for `Collection.find_or_create`"""
  64COLLECTION = "collection"
  65"""`create_type` value for `Collection.find_or_create`"""
  66
  67ChangeList = List[Union[
  68    Tuple[Literal[ADD, DEL], str, 'Collection'],
  69    Tuple[Literal[MOD, TOK], str, 'Collection', 'Collection'],
  70]]
  71ChangeType = Literal[ADD, DEL, MOD, TOK]
  72CollectionItem = Union[ArvadosFile, 'Collection', 'Subcollection']
  73ChangeCallback = Callable[[ChangeType, 'Collection', str, CollectionItem], object]
  74CreateType = Literal[COLLECTION, FILE]
  75Properties = Dict[str, Any]
  76StorageClasses = List[str]
  77
  78class CollectionBase(object):
  79    """Abstract base class for Collection classes
  80
  81    .. ATTENTION:: Internal
  82       This class is meant to be used by other parts of the SDK. User code
  83       should instantiate or subclass `Collection` or one of its subclasses
  84       directly.
  85    """
  86
  87    def __enter__(self):
  88        """Enter a context block with this collection instance"""
  89        return self
  90
  91    def __exit__(self, exc_type, exc_value, traceback):
  92        """Exit a context block with this collection instance"""
  93        pass
  94
  95    def _my_keep(self):
  96        if self._keep_client is None:
  97            self._keep_client = KeepClient(api_client=self._api_client,
  98                                           num_retries=self.num_retries)
  99        return self._keep_client
 100
 101    def stripped_manifest(self) -> str:
 102        """Create a copy of the collection manifest with only size hints
 103
 104        This method returns a string with the current collection's manifest
 105        text with all non-portable locator hints like permission hints and
 106        remote cluster hints removed. The only hints in the returned manifest
 107        will be size hints.
 108        """
 109        raw = self.manifest_text()
 110        clean = []
 111        for line in raw.split("\n"):
 112            fields = line.split()
 113            if fields:
 114                clean_fields = fields[:1] + [
 115                    (re.sub(r'\+[^\d][^\+]*', '', x)
 116                     if re.match(arvados.util.keep_locator_pattern, x)
 117                     else x)
 118                    for x in fields[1:]]
 119                clean += [' '.join(clean_fields), "\n"]
 120        return ''.join(clean)
 121
 122
 123class _WriterFile(_FileLikeObjectBase):
 124    def __init__(self, coll_writer, name):
 125        super(_WriterFile, self).__init__(name, 'wb')
 126        self.dest = coll_writer
 127
 128    def close(self):
 129        super(_WriterFile, self).close()
 130        self.dest.finish_current_file()
 131
 132    @_FileLikeObjectBase._before_close
 133    def write(self, data):
 134        self.dest.write(data)
 135
 136    @_FileLikeObjectBase._before_close
 137    def writelines(self, seq):
 138        for data in seq:
 139            self.write(data)
 140
 141    @_FileLikeObjectBase._before_close
 142    def flush(self):
 143        self.dest.flush_data()
 144
 145
 146class RichCollectionBase(CollectionBase):
 147    """Base class for Collection classes
 148
 149    .. ATTENTION:: Internal
 150       This class is meant to be used by other parts of the SDK. User code
 151       should instantiate or subclass `Collection` or one of its subclasses
 152       directly.
 153    """
 154
 155    def __init__(self, parent=None):
 156        self.parent = parent
 157        self._committed = False
 158        self._has_remote_blocks = False
 159        self._callback = None
 160        self._items = {}
 161
 162    def _my_api(self):
 163        raise NotImplementedError()
 164
 165    def _my_keep(self):
 166        raise NotImplementedError()
 167
 168    def _my_block_manager(self):
 169        raise NotImplementedError()
 170
 171    def writable(self) -> bool:
 172        """Indicate whether this collection object can be modified
 173
 174        This method returns `False` if this object is a `CollectionReader`,
 175        else `True`.
 176        """
 177        raise NotImplementedError()
 178
 179    def root_collection(self) -> 'Collection':
 180        """Get this collection's root collection object
 181
 182        If you open a subcollection with `Collection.find`, calling this method
 183        on that subcollection returns the source Collection object.
 184        """
 185        raise NotImplementedError()
 186
 187    def stream_name(self) -> str:
 188        """Get the name of the manifest stream represented by this collection
 189
 190        If you open a subcollection with `Collection.find`, calling this method
 191        on that subcollection returns the name of the stream you opened.
 192        """
 193        raise NotImplementedError()
 194
 195    @synchronized
 196    def has_remote_blocks(self) -> bool:
 197        """Indiciate whether the collection refers to remote data
 198
 199        Returns `True` if the collection manifest includes any Keep locators
 200        with a remote hint (`+R`), else `False`.
 201        """
 202        if self._has_remote_blocks:
 203            return True
 204        for item in self:
 205            if self[item].has_remote_blocks():
 206                return True
 207        return False
 208
 209    @synchronized
 210    def set_has_remote_blocks(self, val: bool) -> None:
 211        """Cache whether this collection refers to remote blocks
 212
 213        .. ATTENTION:: Internal
 214           This method is only meant to be used by other Collection methods.
 215
 216        Set this collection's cached "has remote blocks" flag to the given
 217        value.
 218        """
 219        self._has_remote_blocks = val
 220        if self.parent:
 221            self.parent.set_has_remote_blocks(val)
 222
 223    @must_be_writable
 224    @synchronized
 225    def find_or_create(
 226            self,
 227            path: str,
 228            create_type: CreateType,
 229    ) -> CollectionItem:
 230        """Get the item at the given path, creating it if necessary
 231
 232        If `path` refers to a stream in this collection, returns a
 233        corresponding `Subcollection` object. If `path` refers to a file in
 234        this collection, returns a corresponding
 235        `arvados.arvfile.ArvadosFile` object. If `path` does not exist in
 236        this collection, then this method creates a new object and returns
 237        it, creating parent streams as needed. The type of object created is
 238        determined by the value of `create_type`.
 239
 240        Arguments:
 241
 242        * path: str --- The path to find or create within this collection.
 243
 244        * create_type: Literal[COLLECTION, FILE] --- The type of object to
 245          create at `path` if one does not exist. Passing `COLLECTION`
 246          creates a stream and returns the corresponding
 247          `Subcollection`. Passing `FILE` creates a new file and returns the
 248          corresponding `arvados.arvfile.ArvadosFile`.
 249        """
 250        pathcomponents = path.split("/", 1)
 251        if pathcomponents[0]:
 252            item = self._items.get(pathcomponents[0])
 253            if len(pathcomponents) == 1:
 254                if item is None:
 255                    # create new file
 256                    if create_type == COLLECTION:
 257                        item = Subcollection(self, pathcomponents[0])
 258                    else:
 259                        item = ArvadosFile(self, pathcomponents[0])
 260                    self._items[pathcomponents[0]] = item
 261                    self.set_committed(False)
 262                    self.notify(ADD, self, pathcomponents[0], item)
 263                return item
 264            else:
 265                if item is None:
 266                    # create new collection
 267                    item = Subcollection(self, pathcomponents[0])
 268                    self._items[pathcomponents[0]] = item
 269                    self.set_committed(False)
 270                    self.notify(ADD, self, pathcomponents[0], item)
 271                if isinstance(item, RichCollectionBase):
 272                    return item.find_or_create(pathcomponents[1], create_type)
 273                else:
 274                    raise IOError(errno.ENOTDIR, "Not a directory", pathcomponents[0])
 275        else:
 276            return self
 277
 278    @synchronized
 279    def find(self, path: str) -> CollectionItem | None:
 280        """Get the item at the given path.
 281
 282        If `path` refers to a stream in this collection, returns a
 283        corresponding `Subcollection` object.
 284
 285        As a special case, if `path` is `"."`, returns the collection itself.
 286
 287        If `path` refers to a file in this collection, returns a corresponding
 288        `arvados.arvfile.ArvadosFile` object.
 289
 290        If `path` does not exist in this collection, then this method returns
 291        `None`.
 292
 293        A `path` that begins with the slash (`/`) character is invalid;
 294        `NotADirectoryError` is raised in this case.
 295
 296        An empty `path` is invalid, and `arvados.errors.ArgumentError` is
 297        raised.
 298
 299        Arguments:
 300
 301        * path: str --- The path to find within this collection.
 302        """
 303        if not path:
 304            raise errors.ArgumentError("Parameter 'path' is empty.")
 305
 306        pathcomponents = path.split("/", 1)
 307        if pathcomponents[0] == '':
 308            raise IOError(errno.ENOTDIR, "Not a directory", pathcomponents[0])
 309
 310        item = self._items.get(pathcomponents[0])
 311        if item is None:
 312            return None
 313        elif len(pathcomponents) == 1:
 314            return item
 315        else:
 316            if isinstance(item, RichCollectionBase):
 317                if pathcomponents[1]:
 318                    return item.find(pathcomponents[1])
 319                else:
 320                    return item
 321            else:
 322                raise IOError(
 323                    errno.ENOTDIR, "Not a directory", pathcomponents[0]
 324                )
 325
 326    @synchronized
 327    def mkdirs(self, path: str) -> 'Subcollection':
 328        """Create and return a subcollection at `path`
 329
 330        If `path` exists within this collection, raises `FileExistsError`.
 331        Otherwise, creates a stream at that path and returns the
 332        corresponding `Subcollection`.
 333        """
 334        if self.find(path) != None:
 335            raise IOError(errno.EEXIST, "Directory or file exists", path)
 336
 337        return self.find_or_create(path, COLLECTION)
 338
 339    def open(
 340            self,
 341            path: str,
 342            mode: str="r",
 343            encoding: Optional[str]=None
 344    ) -> IO:
 345        """Open a file-like object within the collection
 346
 347        This method returns a file-like object that can read and/or write the
 348        file located at `path` within the collection. If you attempt to write
 349        a `path` that does not exist, the file is created with `find_or_create`.
 350        If the file cannot be opened for any other reason, this method raises
 351        `OSError` with an appropriate errno.
 352
 353        Arguments:
 354
 355        * path: str --- The path of the file to open within this collection
 356
 357        * mode: str --- The mode to open this file. Supports all the same
 358          values as `builtins.open`.
 359
 360        * encoding: str | None --- The text encoding of the file. Only used
 361          when the file is opened in text mode. The default is
 362          platform-dependent.
 363
 364        """
 365        if not re.search(r'^[rwa][bt]?\+?$', mode):
 366            raise errors.ArgumentError("Invalid mode {!r}".format(mode))
 367
 368        if mode[0] == 'r' and '+' not in mode:
 369            fclass = ArvadosFileReader
 370            arvfile = self.find(path)
 371        elif not self.writable():
 372            raise IOError(errno.EROFS, "Collection is read only")
 373        else:
 374            fclass = ArvadosFileWriter
 375            arvfile = self.find_or_create(path, FILE)
 376
 377        if arvfile is None:
 378            raise IOError(errno.ENOENT, "File not found", path)
 379        if not isinstance(arvfile, ArvadosFile):
 380            raise IOError(errno.EISDIR, "Is a directory", path)
 381
 382        if mode[0] == 'w':
 383            arvfile.truncate(0)
 384
 385        binmode = mode[0] + 'b' + re.sub('[bt]', '', mode[1:])
 386        f = fclass(arvfile, mode=binmode, num_retries=self.num_retries)
 387        if 'b' not in mode:
 388            bufferclass = io.BufferedRandom if f.writable() else io.BufferedReader
 389            f = io.TextIOWrapper(bufferclass(WrappableFile(f)), encoding=encoding)
 390        return f
 391
 392    def modified(self) -> bool:
 393        """Indicate whether this collection has an API server record
 394
 395        Returns `False` if this collection corresponds to a record loaded from
 396        the API server, `True` otherwise.
 397        """
 398        return not self.committed()
 399
 400    @synchronized
 401    def committed(self):
 402        """Indicate whether this collection has an API server record
 403
 404        Returns `True` if this collection corresponds to a record loaded from
 405        the API server, `False` otherwise.
 406        """
 407        return self._committed
 408
 409    @synchronized
 410    def set_committed(self, value: bool=True):
 411        """Cache whether this collection has an API server record
 412
 413        .. ATTENTION:: Internal
 414           This method is only meant to be used by other Collection methods.
 415
 416        Set this collection's cached "committed" flag to the given
 417        value and propagates it as needed.
 418        """
 419        if value == self._committed:
 420            return
 421        if value:
 422            for k,v in self._items.items():
 423                v.set_committed(True)
 424            self._committed = True
 425        else:
 426            self._committed = False
 427            if self.parent is not None:
 428                self.parent.set_committed(False)
 429
 430    @synchronized
 431    def __iter__(self) -> Iterator[str]:
 432        """Iterate names of streams and files in this collection
 433
 434        This method does not recurse. It only iterates the contents of this
 435        collection's corresponding stream.
 436        """
 437        return iter(self._items)
 438
 439    @synchronized
 440    def __getitem__(self, k: str) -> CollectionItem:
 441        """Get a `arvados.arvfile.ArvadosFile` or `Subcollection` in this collection
 442
 443        This method does not recurse. If you want to search a path, use
 444        `RichCollectionBase.find` instead.
 445        """
 446        return self._items[k]
 447
 448    @synchronized
 449    def __contains__(self, k: str) -> bool:
 450        """Indicate whether this collection has an item with this name
 451
 452        This method does not recurse. It you want to check a path, use
 453        `RichCollectionBase.exists` instead.
 454        """
 455        return k in self._items
 456
 457    @synchronized
 458    def __len__(self):
 459        """Get the number of items directly contained in this collection
 460
 461        This method does not recurse. It only counts the streams and files
 462        in this collection's corresponding stream.
 463        """
 464        return len(self._items)
 465
 466    @must_be_writable
 467    @synchronized
 468    def __delitem__(self, p: str) -> None:
 469        """Delete an item from this collection's stream
 470
 471        This method does not recurse. If you want to remove an item by a
 472        path, use `RichCollectionBase.remove` instead.
 473        """
 474        del self._items[p]
 475        self.set_committed(False)
 476        self.notify(DEL, self, p, None)
 477
 478    @synchronized
 479    def keys(self) -> Iterator[str]:
 480        """Iterate names of streams and files in this collection
 481
 482        This method does not recurse. It only iterates the contents of this
 483        collection's corresponding stream.
 484        """
 485        return self._items.keys()
 486
 487    @synchronized
 488    def values(self) -> List[CollectionItem]:
 489        """Get a list of objects in this collection's stream
 490
 491        The return value includes a `Subcollection` for every stream, and an
 492        `arvados.arvfile.ArvadosFile` for every file, directly within this
 493        collection's stream.  This method does not recurse.
 494        """
 495        return list(self._items.values())
 496
 497    @synchronized
 498    def items(self) -> List[Tuple[str, CollectionItem]]:
 499        """Get a list of `(name, object)` tuples from this collection's stream
 500
 501        The return value includes a `Subcollection` for every stream, and an
 502        `arvados.arvfile.ArvadosFile` for every file, directly within this
 503        collection's stream.  This method does not recurse.
 504        """
 505        return list(self._items.items())
 506
 507    def exists(self, path: str) -> bool:
 508        """Indicate whether this collection includes an item at `path`
 509
 510        This method returns `True` if `path` refers to a stream or file within
 511        this collection, else `False`.
 512
 513        Arguments:
 514
 515        * path: str --- The path to check for existence within this collection
 516        """
 517        return self.find(path) is not None
 518
 519    @must_be_writable
 520    @synchronized
 521    def remove(self, path: str, recursive: bool=False) -> None:
 522        """Remove the file or stream at `path`
 523
 524        Arguments:
 525
 526        * path: str --- The path of the item to remove from the collection
 527
 528        * recursive: bool --- Controls the method's behavior if `path` refers
 529          to a nonempty stream. If `False` (the default), this method raises
 530          `OSError` with errno `ENOTEMPTY`. If `True`, this method removes all
 531          items under the stream.
 532        """
 533        if not path:
 534            raise errors.ArgumentError("Parameter 'path' is empty.")
 535
 536        pathcomponents = path.split("/", 1)
 537        item = self._items.get(pathcomponents[0])
 538        if item is None:
 539            raise IOError(errno.ENOENT, "File not found", path)
 540        if len(pathcomponents) == 1:
 541            if isinstance(self._items[pathcomponents[0]], RichCollectionBase) and len(self._items[pathcomponents[0]]) > 0 and not recursive:
 542                raise IOError(errno.ENOTEMPTY, "Directory not empty", path)
 543            deleteditem = self._items[pathcomponents[0]]
 544            del self._items[pathcomponents[0]]
 545            self.set_committed(False)
 546            self.notify(DEL, self, pathcomponents[0], deleteditem)
 547        else:
 548            item.remove(pathcomponents[1], recursive=recursive)
 549
 550    def _clonefrom(self, source):
 551        for k,v in source.items():
 552            self._items[k] = v.clone(self, k)
 553
 554    def clone(self):
 555        raise NotImplementedError()
 556
 557    @must_be_writable
 558    @synchronized
 559    def add(
 560            self,
 561            source_obj: CollectionItem,
 562            target_name: str,
 563            overwrite: bool=False,
 564            reparent: bool=False,
 565    ) -> None:
 566        """Copy or move a file or subcollection object to this collection
 567
 568        Arguments:
 569
 570        * source_obj: arvados.arvfile.ArvadosFile | Subcollection --- The file or subcollection
 571          to add to this collection
 572
 573        * target_name: str --- The path inside this collection where
 574          `source_obj` should be added.
 575
 576        * overwrite: bool --- Controls the behavior of this method when the
 577          collection already contains an object at `target_name`. If `False`
 578          (the default), this method will raise `FileExistsError`. If `True`,
 579          the object at `target_name` will be replaced with `source_obj`.
 580
 581        * reparent: bool --- Controls whether this method copies or moves
 582          `source_obj`. If `False` (the default), `source_obj` is copied into
 583          this collection. If `True`, `source_obj` is moved into this
 584          collection.
 585        """
 586        if target_name in self and not overwrite:
 587            raise IOError(errno.EEXIST, "File already exists", target_name)
 588
 589        modified_from = None
 590        if target_name in self:
 591            modified_from = self[target_name]
 592
 593        # Actually make the move or copy.
 594        if reparent:
 595            source_obj._reparent(self, target_name)
 596            item = source_obj
 597        else:
 598            item = source_obj.clone(self, target_name)
 599
 600        self._items[target_name] = item
 601        self.set_committed(False)
 602        if not self._has_remote_blocks and source_obj.has_remote_blocks():
 603            self.set_has_remote_blocks(True)
 604
 605        if modified_from:
 606            self.notify(MOD, self, target_name, (modified_from, item))
 607        else:
 608            self.notify(ADD, self, target_name, item)
 609
 610    def _get_src_target(self, source, target_path, source_collection, create_dest):
 611        if source_collection is None:
 612            source_collection = self
 613
 614        # Find the object
 615        if isinstance(source, str):
 616            source_obj = source_collection.find(source)
 617            if source_obj is None:
 618                raise IOError(errno.ENOENT, "File not found", source)
 619            sourcecomponents = source.split("/")
 620        else:
 621            source_obj = source
 622            sourcecomponents = None
 623
 624        # Find parent collection the target path
 625        targetcomponents = target_path.split("/")
 626
 627        # Determine the name to use.
 628        target_name = targetcomponents[-1] if targetcomponents[-1] else sourcecomponents[-1]
 629
 630        if not target_name:
 631            raise errors.ArgumentError("Target path is empty and source is an object.  Cannot determine destination filename to use.")
 632
 633        if create_dest:
 634            target_dir = self.find_or_create("/".join(targetcomponents[0:-1]), COLLECTION)
 635        else:
 636            if len(targetcomponents) > 1:
 637                target_dir = self.find("/".join(targetcomponents[0:-1]))
 638            else:
 639                target_dir = self
 640
 641        if target_dir is None:
 642            raise IOError(errno.ENOENT, "Target directory not found", target_name)
 643
 644        if target_name in target_dir and isinstance(target_dir[target_name], RichCollectionBase) and sourcecomponents:
 645            target_dir = target_dir[target_name]
 646            target_name = sourcecomponents[-1]
 647
 648        return (source_obj, target_dir, target_name)
 649
 650    @must_be_writable
 651    @synchronized
 652    def copy(
 653            self,
 654            source: Union[str, CollectionItem],
 655            target_path: str,
 656            source_collection: Optional['RichCollectionBase']=None,
 657            overwrite: bool=False,
 658    ) -> None:
 659        """Copy a file or subcollection object to this collection
 660
 661        Arguments:
 662
 663        * source: str | arvados.arvfile.ArvadosFile |
 664          arvados.collection.Subcollection --- The file or subcollection to
 665          add to this collection. If `source` is a str, the object will be
 666          found by looking up this path from `source_collection` (see
 667          below).
 668
 669        * target_path: str --- The path inside this collection where the
 670          source object should be added.
 671
 672        * source_collection: arvados.collection.Collection | None --- The
 673          collection to find the source object from when `source` is a
 674          path. Defaults to the current collection (`self`).
 675
 676        * overwrite: bool --- Controls the behavior of this method when the
 677          collection already contains an object at `target_path`. If `False`
 678          (the default), this method will raise `FileExistsError`. If `True`,
 679          the object at `target_path` will be replaced with `source_obj`.
 680        """
 681        source_obj, target_dir, target_name = self._get_src_target(source, target_path, source_collection, True)
 682        target_dir.add(source_obj, target_name, overwrite, False)
 683
 684    @must_be_writable
 685    @synchronized
 686    def rename(
 687            self,
 688            source: Union[str, CollectionItem],
 689            target_path: str,
 690            source_collection: Optional['RichCollectionBase']=None,
 691            overwrite: bool=False,
 692    ) -> None:
 693        """Move a file or subcollection object to this collection
 694
 695        Arguments:
 696
 697        * source: str | arvados.arvfile.ArvadosFile |
 698          arvados.collection.Subcollection --- The file or subcollection to
 699          add to this collection. If `source` is a str, the object will be
 700          found by looking up this path from `source_collection` (see
 701          below).
 702
 703        * target_path: str --- The path inside this collection where the
 704          source object should be added.
 705
 706        * source_collection: arvados.collection.Collection | None --- The
 707          collection to find the source object from when `source` is a
 708          path. Defaults to the current collection (`self`).
 709
 710        * overwrite: bool --- Controls the behavior of this method when the
 711          collection already contains an object at `target_path`. If `False`
 712          (the default), this method will raise `FileExistsError`. If `True`,
 713          the object at `target_path` will be replaced with `source_obj`.
 714        """
 715        source_obj, target_dir, target_name = self._get_src_target(source, target_path, source_collection, False)
 716        if not source_obj.writable():
 717            raise IOError(errno.EROFS, "Source collection is read only", source)
 718        target_dir.add(source_obj, target_name, overwrite, True)
 719
 720    def portable_manifest_text(self, stream_name: str=".") -> str:
 721        """Get the portable manifest text for this collection
 722
 723        The portable manifest text is normalized, and does not include access
 724        tokens. This method does not flush outstanding blocks to Keep.
 725
 726        Arguments:
 727
 728        * stream_name: str --- The name to use for this collection's stream in
 729          the generated manifest. Default `'.'`.
 730        """
 731        return self._get_manifest_text(stream_name, True, True)
 732
 733    @synchronized
 734    def manifest_text(
 735            self,
 736            stream_name: str=".",
 737            strip: bool=False,
 738            normalize: bool=False,
 739            only_committed: bool=False,
 740    ) -> str:
 741        """Get the manifest text for this collection
 742
 743        Arguments:
 744
 745        * stream_name: str --- The name to use for this collection's stream in
 746          the generated manifest. Default `'.'`.
 747
 748        * strip: bool --- Controls whether or not the returned manifest text
 749          includes access tokens. If `False` (the default), the manifest text
 750          will include access tokens. If `True`, the manifest text will not
 751          include access tokens.
 752
 753        * normalize: bool --- Controls whether or not the returned manifest
 754          text is normalized. Default `False`.
 755
 756        * only_committed: bool --- Controls whether or not this method uploads
 757          pending data to Keep before building and returning the manifest text.
 758          If `False` (the default), this method will finish uploading all data
 759          to Keep, then return the final manifest. If `True`, this method will
 760          build and return a manifest that only refers to the data that has
 761          finished uploading at the time this method was called.
 762        """
 763        if not only_committed:
 764            self._my_block_manager().commit_all()
 765        return self._get_manifest_text(stream_name, strip, normalize,
 766                                       only_committed=only_committed)
 767
 768    @synchronized
 769    def _get_manifest_text(self, stream_name, strip, normalize, only_committed=False):
 770        """Get the manifest text for this collection, sub collections and files.
 771
 772        :stream_name:
 773          Name to use for this stream (directory)
 774
 775        :strip:
 776          If True, remove signing tokens from block locators if present.
 777          If False (default), block locators are left unchanged.
 778
 779        :normalize:
 780          If True, always export the manifest text in normalized form
 781          even if the Collection is not modified.  If False (default) and the collection
 782          is not modified, return the original manifest text even if it is not
 783          in normalized form.
 784
 785        :only_committed:
 786          If True, only include blocks that were already committed to Keep.
 787
 788        """
 789
 790        if not self.committed() or self._manifest_text is None or normalize:
 791            stream = {}
 792            buf = []
 793            sorted_keys = sorted(self.keys())
 794            for filename in [s for s in sorted_keys if isinstance(self[s], ArvadosFile)]:
 795                # Create a stream per file `k`
 796                arvfile = self[filename]
 797                filestream = []
 798                for segment in arvfile.segments():
 799                    loc = segment.locator
 800                    if arvfile.parent._my_block_manager().is_bufferblock(loc):
 801                        if only_committed:
 802                            continue
 803                        loc = arvfile.parent._my_block_manager().get_bufferblock(loc).locator()
 804                    if strip:
 805                        loc = KeepLocator(loc).stripped()
 806                    filestream.append(streams.LocatorAndRange(
 807                        loc,
 808                        KeepLocator(loc).size,
 809                        segment.segment_offset,
 810                        segment.range_size,
 811                    ))
 812                stream[filename] = filestream
 813            if stream:
 814                buf.append(" ".join(streams.normalize_stream(stream_name, stream)) + "\n")
 815            for dirname in [s for s in sorted_keys if isinstance(self[s], RichCollectionBase)]:
 816                buf.append(self[dirname].manifest_text(
 817                    stream_name=os.path.join(stream_name, dirname),
 818                    strip=strip, normalize=True, only_committed=only_committed))
 819            return "".join(buf)
 820        else:
 821            if strip:
 822                return self.stripped_manifest()
 823            else:
 824                return self._manifest_text
 825
 826    @synchronized
 827    def _copy_remote_blocks(self, remote_blocks={}):
 828        """Scan through the entire collection and ask Keep to copy remote blocks.
 829
 830        When accessing a remote collection, blocks will have a remote signature
 831        (+R instead of +A). Collect these signatures and request Keep to copy the
 832        blocks to the local cluster, returning local (+A) signatures.
 833
 834        :remote_blocks:
 835          Shared cache of remote to local block mappings. This is used to avoid
 836          doing extra work when blocks are shared by more than one file in
 837          different subdirectories.
 838
 839        """
 840        for item in self:
 841            remote_blocks = self[item]._copy_remote_blocks(remote_blocks)
 842        return remote_blocks
 843
 844    @synchronized
 845    def diff(
 846            self,
 847            end_collection: 'RichCollectionBase',
 848            prefix: str=".",
 849            holding_collection: Optional['Collection']=None,
 850    ) -> ChangeList:
 851        """Build a list of differences between this collection and another
 852
 853        Arguments:
 854
 855        * end_collection: arvados.collection.RichCollectionBase --- A
 856          collection object with the desired end state. The returned diff
 857          list will describe how to go from the current collection object
 858          `self` to `end_collection`.
 859
 860        * prefix: str --- The name to use for this collection's stream in
 861          the diff list. Default `'.'`.
 862
 863        * holding_collection: arvados.collection.Collection | None --- A
 864          collection object used to hold objects for the returned diff
 865          list. By default, a new empty collection is created.
 866        """
 867        changes = []
 868        if holding_collection is None:
 869            holding_collection = Collection(api_client=self._my_api(), keep_client=self._my_keep())
 870        for k in self:
 871            if k not in end_collection:
 872               changes.append((DEL, os.path.join(prefix, k), self[k].clone(holding_collection, "")))
 873        for k in end_collection:
 874            if k in self:
 875                if isinstance(end_collection[k], Subcollection) and isinstance(self[k], Subcollection):
 876                    changes.extend(self[k].diff(end_collection[k], os.path.join(prefix, k), holding_collection))
 877                elif end_collection[k] != self[k]:
 878                    changes.append((MOD, os.path.join(prefix, k), self[k].clone(holding_collection, ""), end_collection[k].clone(holding_collection, "")))
 879                else:
 880                    changes.append((TOK, os.path.join(prefix, k), self[k].clone(holding_collection, ""), end_collection[k].clone(holding_collection, "")))
 881            else:
 882                changes.append((ADD, os.path.join(prefix, k), end_collection[k].clone(holding_collection, "")))
 883        return changes
 884
 885    @must_be_writable
 886    @synchronized
 887    def apply(self, changes: ChangeList) -> None:
 888        """Apply a list of changes from to this collection
 889
 890        This method takes a list of changes generated by
 891        `RichCollectionBase.diff` and applies it to this
 892        collection. Afterward, the state of this collection object will
 893        match the state of `end_collection` passed to `diff`. If a change
 894        conflicts with a local change, it will be saved to an alternate path
 895        indicating the conflict.
 896
 897        Arguments:
 898
 899        * changes: arvados.collection.ChangeList --- The list of differences
 900          generated by `RichCollectionBase.diff`.
 901        """
 902        if changes:
 903            self.set_committed(False)
 904        for change in changes:
 905            event_type = change[0]
 906            path = change[1]
 907            initial = change[2]
 908            local = self.find(path)
 909            conflictpath = "%s~%s~conflict~" % (path, time.strftime("%Y%m%d-%H%M%S",
 910                                                                    time.gmtime()))
 911            if event_type == ADD:
 912                if local is None:
 913                    # No local file at path, safe to copy over new file
 914                    self.copy(initial, path)
 915                elif local is not None and local != initial:
 916                    # There is already local file and it is different:
 917                    # save change to conflict file.
 918                    self.copy(initial, conflictpath)
 919            elif event_type == MOD or event_type == TOK:
 920                final = change[3]
 921                if local == initial:
 922                    # Local matches the "initial" item so it has not
 923                    # changed locally and is safe to update.
 924                    if isinstance(local, ArvadosFile) and isinstance(final, ArvadosFile):
 925                        # Replace contents of local file with new contents
 926                        local.replace_contents(final)
 927                    else:
 928                        # Overwrite path with new item; this can happen if
 929                        # path was a file and is now a collection or vice versa
 930                        self.copy(final, path, overwrite=True)
 931                elif event_type == MOD:
 932                    # Local doesn't match the "start" value or local
 933                    # is missing (presumably deleted) so save change
 934                    # to conflict file.  Don't do this for TOK events
 935                    # which means the file didn't change but only had
 936                    # tokens updated.
 937                    self.copy(final, conflictpath)
 938            elif event_type == DEL:
 939                if local == initial:
 940                    # Local item matches "initial" value, so it is safe to remove.
 941                    self.remove(path, recursive=True)
 942                # else, the file is modified or already removed, in either
 943                # case we don't want to try to remove it.
 944
 945    def portable_data_hash(self) -> str:
 946        """Get the portable data hash for this collection's manifest"""
 947        if self._manifest_locator and self.committed():
 948            # If the collection is already saved on the API server, and it's committed
 949            # then return API server's PDH response.
 950            return self._portable_data_hash
 951        else:
 952            stripped = self.portable_manifest_text().encode()
 953            return '{}+{}'.format(hashlib.md5(stripped).hexdigest(), len(stripped))
 954
 955    @synchronized
 956    def subscribe(self, callback: ChangeCallback) -> None:
 957        """Set a notify callback for changes to this collection
 958
 959        Arguments:
 960
 961        * callback: arvados.collection.ChangeCallback --- The callable to
 962          call each time the collection is changed.
 963        """
 964        if self._callback is None:
 965            self._callback = callback
 966        else:
 967            raise errors.ArgumentError("A callback is already set on this collection.")
 968
 969    @synchronized
 970    def unsubscribe(self) -> None:
 971        """Remove any notify callback set for changes to this collection"""
 972        if self._callback is not None:
 973            self._callback = None
 974
 975    @synchronized
 976    def notify(
 977            self,
 978            event: ChangeType,
 979            collection: 'RichCollectionBase',
 980            name: str,
 981            item: CollectionItem,
 982    ) -> None:
 983        """Notify any subscribed callback about a change to this collection
 984
 985        .. ATTENTION:: Internal
 986           This method is only meant to be used by other Collection methods.
 987
 988        If a callback has been registered with `RichCollectionBase.subscribe`,
 989        it will be called with information about a change to this collection.
 990        Then this notification will be propagated to this collection's root.
 991
 992        Arguments:
 993
 994        * event: Literal[ADD, DEL, MOD, TOK] --- The type of modification to
 995          the collection.
 996
 997        * collection: arvados.collection.RichCollectionBase --- The
 998          collection that was modified.
 999
1000        * name: str --- The name of the file or stream within `collection` that
1001          was modified.
1002
1003        * item: arvados.arvfile.ArvadosFile |
1004          arvados.collection.Subcollection --- For ADD events, the new
1005          contents at `name` within `collection`; for DEL events, the
1006          item that was removed.  For MOD and TOK events, a 2-tuple of
1007          the previous item and the new item (may be the same object
1008          or different, depending on whether the action involved it
1009          being modified in place or replaced).
1010
1011        """
1012        if self._callback:
1013            self._callback(event, collection, name, item)
1014        self.root_collection().notify(event, collection, name, item)
1015
1016    @synchronized
1017    def __eq__(self, other: Any) -> bool:
1018        """Indicate whether this collection object is equal to another"""
1019        if other is self:
1020            return True
1021        if not isinstance(other, RichCollectionBase):
1022            return False
1023        if len(self._items) != len(other):
1024            return False
1025        for k in self._items:
1026            if k not in other:
1027                return False
1028            if self._items[k] != other[k]:
1029                return False
1030        return True
1031
1032    def __ne__(self, other: Any) -> bool:
1033        """Indicate whether this collection object is not equal to another"""
1034        return not self.__eq__(other)
1035
1036    @synchronized
1037    def flush(self) -> None:
1038        """Upload any pending data to Keep"""
1039        for e in self.values():
1040            e.flush()
1041
1042
1043class Collection(RichCollectionBase):
1044    """Read and manipulate an Arvados collection
1045
1046    This class provides a high-level interface to create, read, and update
1047    Arvados collections and their contents. Refer to the Arvados Python SDK
1048    cookbook for [an introduction to using the Collection class][cookbook].
1049
1050    [cookbook]: https://doc.arvados.org/sdk/python/cookbook.html#working-with-collections
1051    """
1052
1053    def __init__(self, manifest_locator_or_text: Optional[str]=None,
1054                 api_client: Optional['arvados.api_resources.ArvadosAPIClient']=None,
1055                 keep_client: Optional['arvados.keep.KeepClient']=None,
1056                 num_retries: int=10,
1057                 parent: Optional['Collection']=None,
1058                 apiconfig: Optional[Mapping[str, str]]=None,
1059                 block_manager: Optional['arvados.arvfile._BlockManager']=None,
1060                 replication_desired: Optional[int]=None,
1061                 storage_classes_desired: Optional[List[str]]=None,
1062                 put_threads: Optional[int]=None):
1063        """Initialize a Collection object
1064
1065        Arguments:
1066
1067        * manifest_locator_or_text: str | None --- This string can contain a
1068          collection manifest text, portable data hash, or UUID. When given a
1069          portable data hash or UUID, this instance will load a collection
1070          record from the API server. Otherwise, this instance will represent a
1071          new collection without an API server record. The default value `None`
1072          instantiates a new collection with an empty manifest.
1073
1074        * api_client: arvados.api_resources.ArvadosAPIClient | None --- The
1075          Arvados API client object this instance uses to make requests. If
1076          none is given, this instance creates its own client using the
1077          settings from `apiconfig` (see below). If your client instantiates
1078          many Collection objects, you can help limit memory utilization by
1079          calling `arvados.api.api` to construct an
1080          `arvados.api.ThreadSafeAPIClient`, and use that as the `api_client`
1081          for every Collection.
1082
1083        * keep_client: arvados.keep.KeepClient | None --- The Keep client
1084          object this instance uses to make requests. If none is given, this
1085          instance creates its own client using its `api_client`.
1086
1087        * num_retries: int --- The number of times that client requests are
1088          retried. Default 10.
1089
1090        * parent: arvados.collection.Collection | None --- The parent Collection
1091          object of this instance, if any. This argument is primarily used by
1092          other Collection methods; user client code shouldn't need to use it.
1093
1094        * apiconfig: Mapping[str, str] | None --- A mapping with entries for
1095          `ARVADOS_API_HOST`, `ARVADOS_API_TOKEN`, and optionally
1096          `ARVADOS_API_HOST_INSECURE`. When no `api_client` is provided, the
1097          Collection object constructs one from these settings. If no
1098          mapping is provided, calls `arvados.config.settings` to get these
1099          parameters from user configuration.
1100
1101        * block_manager: arvados.arvfile._BlockManager | None --- The
1102          _BlockManager object used by this instance to coordinate reading
1103          and writing Keep data blocks. If none is given, this instance
1104          constructs its own. This argument is primarily used by other
1105          Collection methods; user client code shouldn't need to use it.
1106
1107        * replication_desired: int | None --- This controls both the value of
1108          the `replication_desired` field on API collection records saved by
1109          this class, as well as the number of Keep services that the object
1110          writes new data blocks to. If none is given, uses the default value
1111          configured for the cluster.
1112
1113        * storage_classes_desired: list[str] | None --- This controls both
1114          the value of the `storage_classes_desired` field on API collection
1115          records saved by this class, as well as selecting which specific
1116          Keep services the object writes new data blocks to. If none is
1117          given, defaults to an empty list.
1118
1119        * put_threads: int | None --- The number of threads to run
1120          simultaneously to upload data blocks to Keep. This value is used when
1121          building a new `block_manager`. It is unused when a `block_manager`
1122          is provided.
1123        """
1124
1125        if storage_classes_desired and type(storage_classes_desired) is not list:
1126            raise errors.ArgumentError("storage_classes_desired must be list type.")
1127
1128        super(Collection, self).__init__(parent)
1129        self._api_client = api_client
1130        self._keep_client = keep_client
1131
1132        # Use the keep client from ThreadSafeAPIClient
1133        if self._keep_client is None and isinstance(self._api_client, ThreadSafeAPIClient):
1134            self._keep_client = self._api_client.keep
1135
1136        self._block_manager = block_manager
1137        self.replication_desired = replication_desired
1138        self._storage_classes_desired = storage_classes_desired
1139        self.put_threads = put_threads
1140
1141        if apiconfig:
1142            self._config = apiconfig
1143        else:
1144            self._config = config.settings()
1145
1146        self.num_retries = num_retries
1147        self._manifest_locator = None
1148        self._manifest_text = None
1149        self._portable_data_hash = None
1150        self._api_response = None
1151        self._token_refresh_timestamp = 0
1152
1153        self.lock = threading.RLock()
1154        self.events = None
1155
1156        if manifest_locator_or_text:
1157            if re.match(arvados.util.keep_locator_pattern, manifest_locator_or_text):
1158                self._manifest_locator = manifest_locator_or_text
1159            elif re.match(arvados.util.collection_uuid_pattern, manifest_locator_or_text):
1160                self._manifest_locator = manifest_locator_or_text
1161                if not self._has_local_collection_uuid():
1162                    self._has_remote_blocks = True
1163            elif re.match(arvados.util.manifest_pattern, manifest_locator_or_text):
1164                self._manifest_text = manifest_locator_or_text
1165                if '+R' in self._manifest_text:
1166                    self._has_remote_blocks = True
1167            else:
1168                raise errors.ArgumentError(
1169                    "Argument to CollectionReader is not a manifest or a collection UUID")
1170
1171            try:
1172                self._populate()
1173            except errors.SyntaxError as e:
1174                raise errors.ArgumentError("Error processing manifest text: %s", str(e)) from None
1175
1176    def storage_classes_desired(self) -> List[str]:
1177        """Get this collection's `storage_classes_desired` value"""
1178        return self._storage_classes_desired or []
1179
1180    def root_collection(self) -> 'Collection':
1181        return self
1182
1183    def get_properties(self) -> Properties:
1184        """Get this collection's properties
1185
1186        This method always returns a dict. If this collection object does not
1187        have an associated API record, or that record does not have any
1188        properties set, this method returns an empty dict.
1189        """
1190        if self._api_response and self._api_response["properties"]:
1191            return self._api_response["properties"]
1192        else:
1193            return {}
1194
1195    def get_trash_at(self) -> Optional[datetime.datetime]:
1196        """Get this collection's `trash_at` field
1197
1198        This method parses the `trash_at` field of the collection's API
1199        record and returns a datetime from it. If that field is not set, or
1200        this collection object does not have an associated API record,
1201        returns None.
1202        """
1203        if self._api_response and self._api_response["trash_at"]:
1204            try:
1205                return ciso8601.parse_datetime(self._api_response["trash_at"])
1206            except ValueError:
1207                return None
1208        else:
1209            return None
1210
1211    def stream_name(self) -> str:
1212        return "."
1213
1214    def writable(self) -> bool:
1215        return True
1216
1217    @synchronized
1218    @retry_method
1219    def update(
1220            self,
1221            other: Optional['Collection']=None,
1222            num_retries: Optional[int]=None,
1223    ) -> None:
1224        """Merge another collection's contents into this one
1225
1226        This method compares the manifest of this collection instance with
1227        another, then updates this instance's manifest with changes from the
1228        other, renaming files to flag conflicts where necessary.
1229
1230        When called without any arguments, this method reloads the collection's
1231        API record, and updates this instance with any changes that have
1232        appeared server-side. If this instance does not have a corresponding
1233        API record, this method raises `arvados.errors.ArgumentError`.
1234
1235        Arguments:
1236
1237        * other: arvados.collection.Collection | None --- The collection
1238          whose contents should be merged into this instance. When not
1239          provided, this method reloads this collection's API record and
1240          constructs a Collection object from it.  If this instance does not
1241          have a corresponding API record, this method raises
1242          `arvados.errors.ArgumentError`.
1243
1244        * num_retries: int | None --- The number of times to retry reloading
1245          the collection's API record from the API server. If not specified,
1246          uses the `num_retries` provided when this instance was constructed.
1247        """
1248
1249        token_refresh_period = 60*60
1250        time_since_last_token_refresh = (time.time() - self._token_refresh_timestamp)
1251        upstream_response = None
1252
1253        if other is None:
1254            if self._manifest_locator is None:
1255                raise errors.ArgumentError("`other` is None but collection does not have a manifest_locator uuid")
1256
1257            if re.match(arvados.util.portable_data_hash_pattern, self._manifest_locator) and time_since_last_token_refresh < token_refresh_period:
1258                return
1259
1260            upstream_response = self._my_api().collections().get(uuid=self._manifest_locator).execute(num_retries=num_retries)
1261            other = CollectionReader(upstream_response["manifest_text"], self._my_api())
1262
1263        if self.committed():
1264            # 1st case, no local changes, content is the same
1265            if self.portable_data_hash() == other.portable_data_hash() and time_since_last_token_refresh < token_refresh_period:
1266                # No difference in content.  Remember the API record
1267                # (metadata such as name or properties may have changed)
1268                # but don't update the token refresh timestamp.
1269                if upstream_response is not None:
1270                    self._remember_api_response(upstream_response)
1271                return
1272
1273            # 2nd case, no local changes, but either upstream changed
1274            # or we want to refresh tokens.
1275
1276            self.apply(self.diff(other))
1277            if upstream_response is not None:
1278                self._remember_api_response(upstream_response)
1279            self._update_token_timestamp()
1280            self.set_committed(True)
1281            return
1282
1283        # 3rd case, upstream changed, but we also have uncommitted
1284        # changes that we want to incorporate so they don't get lost.
1285
1286        # _manifest_text stores the text from last time we received a
1287        # record from the API server.  This is the state of the
1288        # collection before our uncommitted changes.
1289        baseline = Collection(self._manifest_text, self._my_api())
1290
1291        # Get the set of changes between our baseline and the other
1292        # collection and apply them to self.
1293        #
1294        # If a file was modified in both 'self' and 'other', the
1295        # 'apply' method keeps the contents of 'self' and creates a
1296        # conflict file with the contents of 'other'.
1297        self.apply(baseline.diff(other))
1298
1299        # Remember the new baseline, changes to a file
1300        if upstream_response is not None:
1301            self._remember_api_response(upstream_response)
1302
1303
1304    @synchronized
1305    def _my_api(self):
1306        if self._api_client is None:
1307            self._api_client = ThreadSafeAPIClient(self._config, version='v1')
1308            if self._keep_client is None:
1309                self._keep_client = self._api_client.keep
1310        return self._api_client
1311
1312    @synchronized
1313    def _my_keep(self):
1314        if self._keep_client is None:
1315            if self._api_client is None:
1316                self._my_api()
1317            else:
1318                self._keep_client = KeepClient(api_client=self._api_client)
1319        return self._keep_client
1320
1321    @synchronized
1322    def _my_block_manager(self):
1323        if self._block_manager is None:
1324            copies = (self.replication_desired or
1325                      self._my_api()._rootDesc.get('defaultCollectionReplication',
1326                                                   2))
1327            self._block_manager = _BlockManager(self._my_keep(),
1328                                                copies=copies,
1329                                                put_threads=self.put_threads,
1330                                                num_retries=self.num_retries,
1331                                                storage_classes_func=self.storage_classes_desired)
1332        return self._block_manager
1333
1334    def _remember_api_response(self, response):
1335        self._api_response = response
1336        self._manifest_text = self._api_response['manifest_text']
1337        self._portable_data_hash = self._api_response['portable_data_hash']
1338
1339    def _update_token_timestamp(self):
1340        self._token_refresh_timestamp = time.time()
1341
1342    def _populate_from_api_server(self):
1343        # As in KeepClient itself, we must wait until the last
1344        # possible moment to instantiate an API client, in order to
1345        # avoid tripping up clients that don't have access to an API
1346        # server.  If we do build one, make sure our Keep client uses
1347        # it.  If instantiation fails, we'll fall back to the except
1348        # clause, just like any other Collection lookup
1349        # failure. Return an exception, or None if successful.
1350        self._remember_api_response(self._my_api().collections().get(
1351            uuid=self._manifest_locator).execute(
1352                num_retries=self.num_retries))
1353
1354        # If not overriden via kwargs, we should try to load the
1355        # replication_desired and storage_classes_desired from the API server
1356        if self.replication_desired is None:
1357            self.replication_desired = self._api_response.get('replication_desired', None)
1358        if self._storage_classes_desired is None:
1359            self._storage_classes_desired = self._api_response.get('storage_classes_desired', None)
1360
1361    def _populate(self):
1362        if self._manifest_text is None:
1363            if self._manifest_locator is None:
1364                return
1365            else:
1366                self._populate_from_api_server()
1367        self._baseline_manifest = self._manifest_text
1368        self._import_manifest(self._manifest_text)
1369
1370    def _has_collection_uuid(self):
1371        return self._manifest_locator is not None and re.match(arvados.util.collection_uuid_pattern, self._manifest_locator)
1372
1373    def _has_local_collection_uuid(self):
1374        return self._has_collection_uuid and \
1375            self._my_api()._rootDesc['uuidPrefix'] == self._manifest_locator.split('-')[0]
1376
1377    def __enter__(self):
1378        return self
1379
1380    def __exit__(self, exc_type, exc_value, traceback):
1381        """Exit a context with this collection instance
1382
1383        If no exception was raised inside the context block, and this
1384        collection is writable and has a corresponding API record, that
1385        record will be updated to match the state of this instance at the end
1386        of the block.
1387        """
1388        if exc_type is None:
1389            if self.writable() and self._has_collection_uuid():
1390                self.save()
1391        self.stop_threads()
1392
1393    def stop_threads(self) -> None:
1394        """Stop background Keep upload/download threads"""
1395        if self._block_manager is not None:
1396            self._block_manager.stop_threads()
1397
1398    @synchronized
1399    def manifest_locator(self) -> Optional[str]:
1400        """Get this collection's manifest locator, if any
1401
1402        * If this collection instance is associated with an API record with a
1403          UUID, return that.
1404        * Otherwise, if this collection instance was loaded from an API record
1405          by portable data hash, return that.
1406        * Otherwise, return `None`.
1407        """
1408        return self._manifest_locator
1409
1410    @synchronized
1411    def clone(
1412            self,
1413            new_parent: Optional['Collection']=None,
1414            new_name: Optional[str]=None,
1415            readonly: bool=False,
1416            new_config: Optional[Mapping[str, str]]=None,
1417    ) -> 'Collection':
1418        """Create a Collection object with the same contents as this instance
1419
1420        This method creates a new Collection object with contents that match
1421        this instance's. The new collection will not be associated with any API
1422        record.
1423
1424        Arguments:
1425
1426        * new_parent: arvados.collection.Collection | None --- This value is
1427          passed to the new Collection's constructor as the `parent`
1428          argument.
1429
1430        * new_name: str | None --- This value is unused.
1431
1432        * readonly: bool --- If this value is true, this method constructs and
1433          returns a `CollectionReader`. Otherwise, it returns a mutable
1434          `Collection`. Default `False`.
1435
1436        * new_config: Mapping[str, str] | None --- This value is passed to the
1437          new Collection's constructor as `apiconfig`. If no value is provided,
1438          defaults to the configuration passed to this instance's constructor.
1439        """
1440        if new_config is None:
1441            new_config = self._config
1442        if readonly:
1443            newcollection = CollectionReader(parent=new_parent, apiconfig=new_config)
1444        else:
1445            newcollection = Collection(parent=new_parent, apiconfig=new_config)
1446
1447        newcollection._clonefrom(self)
1448        return newcollection
1449
1450    @synchronized
1451    def api_response(self) -> Optional[Dict[str, Any]]:
1452        """Get this instance's associated API record
1453
1454        If this Collection instance has an associated API record, return it.
1455        Otherwise, return `None`.
1456        """
1457        return self._api_response
1458
1459    def find_or_create(
1460            self,
1461            path: str,
1462            create_type: CreateType,
1463    ) -> CollectionItem:
1464        if path == ".":
1465            return self
1466        else:
1467            return super(Collection, self).find_or_create(path[2:] if path.startswith("./") else path, create_type)
1468
1469    def find(self, path: str) -> CollectionItem | None:
1470        if path == ".":
1471            return self
1472        else:
1473            return super(Collection, self).find(path[2:] if path.startswith("./") else path)
1474
1475    def remove(self, path: str, recursive: bool=False) -> None:
1476        if path == ".":
1477            raise errors.ArgumentError("Cannot remove '.'")
1478        else:
1479            return super(Collection, self).remove(path[2:] if path.startswith("./") else path, recursive)
1480
1481    @must_be_writable
1482    @synchronized
1483    @retry_method
1484    def save(
1485            self,
1486            properties: Optional[Properties]=None,
1487            storage_classes: Optional[StorageClasses]=None,
1488            trash_at: Optional[datetime.datetime]=None,
1489            merge: bool=True,
1490            num_retries: Optional[int]=None,
1491            preserve_version: bool=False,
1492    ) -> str:
1493        """Save collection to an existing API record
1494
1495        This method updates the instance's corresponding API record to match
1496        the instance's state. If this instance does not have a corresponding API
1497        record yet, raises `AssertionError`. (To create a new API record, use
1498        `Collection.save_new`.) This method returns the saved collection
1499        manifest.
1500
1501        Arguments:
1502
1503        * properties: dict[str, Any] | None --- If provided, the API record will
1504          be updated with these properties. Note this will completely replace
1505          any existing properties.
1506
1507        * storage_classes: list[str] | None --- If provided, the API record will
1508          be updated with this value in the `storage_classes_desired` field.
1509          This value will also be saved on the instance and used for any
1510          changes that follow.
1511
1512        * trash_at: datetime.datetime | None --- If provided, the API record
1513          will be updated with this value in the `trash_at` field.
1514
1515        * merge: bool --- If `True` (the default), this method will first
1516          reload this collection's API record, and merge any new contents into
1517          this instance before saving changes. See `Collection.update` for
1518          details.
1519
1520        * num_retries: int | None --- The number of times to retry reloading
1521          the collection's API record from the API server. If not specified,
1522          uses the `num_retries` provided when this instance was constructed.
1523
1524        * preserve_version: bool --- This value will be passed to directly
1525          to the underlying API call. If `True`, the Arvados API will
1526          preserve the versions of this collection both immediately before
1527          and after the update. If `True` when the API server is not
1528          configured with collection versioning, this method raises
1529          `arvados.errors.ArgumentError`.
1530        """
1531        if properties and type(properties) is not dict:
1532            raise errors.ArgumentError("properties must be dictionary type.")
1533
1534        if storage_classes and type(storage_classes) is not list:
1535            raise errors.ArgumentError("storage_classes must be list type.")
1536        if storage_classes:
1537            self._storage_classes_desired = storage_classes
1538
1539        if trash_at and type(trash_at) is not datetime.datetime:
1540            raise errors.ArgumentError("trash_at must be datetime type.")
1541
1542        if preserve_version and not self._my_api().config()['Collections'].get('CollectionVersioning', False):
1543            raise errors.ArgumentError("preserve_version is not supported when CollectionVersioning is not enabled.")
1544
1545        body={}
1546        if properties:
1547            body["properties"] = properties
1548        if self.storage_classes_desired():
1549            body["storage_classes_desired"] = self.storage_classes_desired()
1550        if trash_at:
1551            t = trash_at.strftime("%Y-%m-%dT%H:%M:%S.%fZ")
1552            body["trash_at"] = t
1553        if preserve_version:
1554            body["preserve_version"] = preserve_version
1555
1556        if not self.committed():
1557            if self._has_remote_blocks:
1558                # Copy any remote blocks to the local cluster.
1559                self._copy_remote_blocks(remote_blocks={})
1560                self._has_remote_blocks = False
1561            if not self._has_collection_uuid():
1562                raise AssertionError("Collection manifest_locator is not a collection uuid.  Use save_new() for new collections.")
1563            elif not self._has_local_collection_uuid():
1564                raise AssertionError("Collection manifest_locator is from a remote cluster. Use save_new() to save it on the local cluster.")
1565
1566            self._my_block_manager().commit_all()
1567
1568            if merge:
1569                self.update()
1570
1571            text = self.manifest_text(strip=False)
1572            body['manifest_text'] = text
1573
1574            self._remember_api_response(self._my_api().collections().update(
1575                uuid=self._manifest_locator,
1576                body=body
1577                ).execute(num_retries=num_retries))
1578            self.set_committed(True)
1579        elif body:
1580            self._remember_api_response(self._my_api().collections().update(
1581                uuid=self._manifest_locator,
1582                body=body
1583                ).execute(num_retries=num_retries))
1584
1585        return self._manifest_text
1586
1587
1588    @must_be_writable
1589    @synchronized
1590    @retry_method
1591    def save_new(
1592            self,
1593            name: Optional[str]=None,
1594            create_collection_record: bool=True,
1595            owner_uuid: Optional[str]=None,
1596            properties: Optional[Properties]=None,
1597            storage_classes: Optional[StorageClasses]=None,
1598            trash_at: Optional[datetime.datetime]=None,
1599            ensure_unique_name: bool=False,
1600            num_retries: Optional[int]=None,
1601            preserve_version: bool=False,
1602    ):
1603        """Save collection to a new API record
1604
1605        This method finishes uploading new data blocks and (optionally)
1606        creates a new API collection record with the provided data. If a new
1607        record is created, this instance becomes associated with that record
1608        for future updates like `save()`. This method returns the saved
1609        collection manifest.
1610
1611        Arguments:
1612
1613        * name: str | None --- The `name` field to use on the new collection
1614          record. If not specified, a generic default name is generated.
1615
1616        * create_collection_record: bool --- If `True` (the default), creates a
1617          collection record on the API server. If `False`, the method finishes
1618          all data uploads and only returns the resulting collection manifest
1619          without sending it to the API server.
1620
1621        * owner_uuid: str | None --- The `owner_uuid` field to use on the
1622          new collection record.
1623
1624        * properties: dict[str, Any] | None --- The `properties` field to use on
1625          the new collection record.
1626
1627        * storage_classes: list[str] | None --- The
1628          `storage_classes_desired` field to use on the new collection record.
1629
1630        * trash_at: datetime.datetime | None --- The `trash_at` field to use
1631          on the new collection record.
1632
1633        * ensure_unique_name: bool --- This value is passed directly to the
1634          Arvados API when creating the collection record. If `True`, the API
1635          server may modify the submitted `name` to ensure the collection's
1636          `name`+`owner_uuid` combination is unique. If `False` (the default),
1637          if a collection already exists with this same `name`+`owner_uuid`
1638          combination, creating a collection record will raise a validation
1639          error.
1640
1641        * num_retries: int | None --- The number of times to retry reloading
1642          the collection's API record from the API server. If not specified,
1643          uses the `num_retries` provided when this instance was constructed.
1644
1645        * preserve_version: bool --- This value will be passed to directly
1646          to the underlying API call. If `True`, the Arvados API will
1647          preserve the versions of this collection both immediately before
1648          and after the update. If `True` when the API server is not
1649          configured with collection versioning, this method raises
1650          `arvados.errors.ArgumentError`.
1651        """
1652        if properties and type(properties) is not dict:
1653            raise errors.ArgumentError("properties must be dictionary type.")
1654
1655        if storage_classes and type(storage_classes) is not list:
1656            raise errors.ArgumentError("storage_classes must be list type.")
1657
1658        if trash_at and type(trash_at) is not datetime.datetime:
1659            raise errors.ArgumentError("trash_at must be datetime type.")
1660
1661        if preserve_version and not self._my_api().config()['Collections'].get('CollectionVersioning', False):
1662            raise errors.ArgumentError("preserve_version is not supported when CollectionVersioning is not enabled.")
1663
1664        if self._has_remote_blocks:
1665            # Copy any remote blocks to the local cluster.
1666            self._copy_remote_blocks(remote_blocks={})
1667            self._has_remote_blocks = False
1668
1669        if storage_classes:
1670            self._storage_classes_desired = storage_classes
1671
1672        self._my_block_manager().commit_all()
1673        text = self.manifest_text(strip=False)
1674
1675        if create_collection_record:
1676            if name is None:
1677                name = "New collection"
1678                ensure_unique_name = True
1679
1680            body = {"manifest_text": text,
1681                    "name": name,
1682                    "replication_desired": self.replication_desired}
1683            if owner_uuid:
1684                body["owner_uuid"] = owner_uuid
1685            if properties:
1686                body["properties"] = properties
1687            if self.storage_classes_desired():
1688                body["storage_classes_desired"] = self.storage_classes_desired()
1689            if trash_at:
1690                t = trash_at.strftime("%Y-%m-%dT%H:%M:%S.%fZ")
1691                body["trash_at"] = t
1692            if preserve_version:
1693                body["preserve_version"] = preserve_version
1694
1695            self._remember_api_response(self._my_api().collections().create(ensure_unique_name=ensure_unique_name, body=body).execute(num_retries=num_retries))
1696            self._manifest_locator = self._api_response["uuid"]
1697            self.set_committed(True)
1698
1699        return text
1700
1701    _token_re = re.compile(r'(\S+)(\s+|$)')
1702    _block_re = re.compile(r'[0-9a-f]{32}\+(\d+)(\+\S+)*')
1703    _segment_re = re.compile(r'(\d+):(\d+):(\S+)')
1704
1705    def _unescape_manifest_path(self, path):
1706        return re.sub(r'\\([0-3][0-7][0-7])', lambda m: chr(int(m.group(1), 8)), path)
1707
1708    @synchronized
1709    def _import_manifest(self, manifest_text):
1710        """Import a manifest into a `Collection`.
1711
1712        :manifest_text:
1713          The manifest text to import from.
1714
1715        """
1716        if len(self) > 0:
1717            raise ArgumentError("Can only import manifest into an empty collection")
1718
1719        STREAM_NAME = 0
1720        BLOCKS = 1
1721        SEGMENTS = 2
1722
1723        stream_name = None
1724        state = STREAM_NAME
1725
1726        for token_and_separator in self._token_re.finditer(manifest_text):
1727            tok = token_and_separator.group(1)
1728            sep = token_and_separator.group(2)
1729
1730            if state == STREAM_NAME:
1731                # starting a new stream
1732                stream_name = self._unescape_manifest_path(tok)
1733                blocks = []
1734                segments = []
1735                streamoffset = 0
1736                state = BLOCKS
1737                self.find_or_create(stream_name, COLLECTION)
1738                continue
1739
1740            if state == BLOCKS:
1741                block_locator = self._block_re.match(tok)
1742                if block_locator:
1743                    blocksize = int(block_locator.group(1))
1744                    blocks.append(streams.Range(tok, streamoffset, blocksize, 0))
1745                    streamoffset += blocksize
1746                else:
1747                    state = SEGMENTS
1748
1749            if state == SEGMENTS:
1750                file_segment = self._segment_re.match(tok)
1751                if file_segment:
1752                    pos = int(file_segment.group(1))
1753                    size = int(file_segment.group(2))
1754                    name = self._unescape_manifest_path(file_segment.group(3))
1755                    if name.split('/')[-1] == '.':
1756                        # placeholder for persisting an empty directory, not a real file
1757                        if len(name) > 2:
1758                            self.find_or_create(os.path.join(stream_name, name[:-2]), COLLECTION)
1759                    else:
1760                        filepath = os.path.join(stream_name, name)
1761                        try:
1762                            afile = self.find_or_create(filepath, FILE)
1763                        except IOError as e:
1764                            if e.errno == errno.ENOTDIR:
1765                                raise errors.SyntaxError("Dir part of %s conflicts with file of the same name.", filepath) from None
1766                            else:
1767                                raise e from None
1768                        if isinstance(afile, ArvadosFile):
1769                            afile.add_segment(blocks, pos, size)
1770                        else:
1771                            raise errors.SyntaxError("File %s conflicts with stream of the same name.", filepath)
1772                else:
1773                    # error!
1774                    raise errors.SyntaxError("Invalid manifest format, expected file segment but did not match format: '%s'" % tok)
1775
1776            if sep == "\n":
1777                stream_name = None
1778                state = STREAM_NAME
1779
1780        self._update_token_timestamp()
1781        self.set_committed(True)
1782
1783    @synchronized
1784    def notify(
1785            self,
1786            event: ChangeType,
1787            collection: 'RichCollectionBase',
1788            name: str,
1789            item: CollectionItem,
1790    ) -> None:
1791        if self._callback:
1792            self._callback(event, collection, name, item)
1793
1794
1795class Subcollection(RichCollectionBase):
1796    """Read and manipulate a stream/directory within an Arvados collection
1797
1798    This class represents a single stream (like a directory) within an Arvados
1799    `Collection`. It is returned by `Collection.find` and provides the same API.
1800    Operations that work on the API collection record propagate to the parent
1801    `Collection` object.
1802    """
1803
1804    def __init__(self, parent, name):
1805        super(Subcollection, self).__init__(parent)
1806        self.lock = self.root_collection().lock
1807        self._manifest_text = None
1808        self.name = name
1809        self.num_retries = parent.num_retries
1810
1811    def root_collection(self) -> 'Collection':
1812        return self.parent.root_collection()
1813
1814    def writable(self) -> bool:
1815        return self.root_collection().writable()
1816
1817    def _my_api(self):
1818        return self.root_collection()._my_api()
1819
1820    def _my_keep(self):
1821        return self.root_collection()._my_keep()
1822
1823    def _my_block_manager(self):
1824        return self.root_collection()._my_block_manager()
1825
1826    def stream_name(self) -> str:
1827        return os.path.join(self.parent.stream_name(), self.name)
1828
1829    @synchronized
1830    def clone(
1831            self,
1832            new_parent: Optional['Collection']=None,
1833            new_name: Optional[str]=None,
1834    ) -> 'Subcollection':
1835        c = Subcollection(new_parent, new_name)
1836        c._clonefrom(self)
1837        return c
1838
1839    @must_be_writable
1840    @synchronized
1841    def _reparent(self, newparent, newname):
1842        self.set_committed(False)
1843        self.flush()
1844        self.parent.remove(self.name, recursive=True)
1845        self.parent = newparent
1846        self.name = newname
1847        self.lock = self.parent.root_collection().lock
1848
1849    @synchronized
1850    def _get_manifest_text(self, stream_name, strip, normalize, only_committed=False):
1851        """Encode empty directories by using an \056-named (".") empty file"""
1852        if len(self._items) == 0:
1853            return "%s %s 0:0:\\056\n" % (
1854                streams.escape(stream_name), config.EMPTY_BLOCK_LOCATOR)
1855        return super(Subcollection, self)._get_manifest_text(stream_name,
1856                                                             strip, normalize,
1857                                                             only_committed)
1858
1859
1860class CollectionReader(Collection):
1861    """Read-only `Collection` subclass
1862
1863    This class will never create or update any API collection records. You can
1864    use this class for additional code safety when you only need to read
1865    existing collections.
1866    """
1867    def __init__(self, manifest_locator_or_text, *args, **kwargs):
1868        self._in_init = True
1869        super(CollectionReader, self).__init__(manifest_locator_or_text, *args, **kwargs)
1870        self._in_init = False
1871
1872        # Forego any locking since it should never change once initialized.
1873        self.lock = NoopLock()
1874
1875        # Backwards compatability with old CollectionReader
1876        # all_streams() and all_files()
1877        self._streams = None
1878
1879    def writable(self) -> bool:
1880        return self._in_init