def consolidate_metadata(store: StoreLike, metadata_key=".zmetadata"):
    """
    Consolidate all metadata for groups and arrays within the given store
    into a single resource and put it under the given key.

    This produces a single object in the backend store, containing all the
    metadata read from all the zarr-related keys that can be found. After
    metadata have been consolidated, use :func:`open_consolidated` to open
    the root group in optimised, read-only mode, using the consolidated
    metadata to reduce the number of read operations on the backend store.

    Note, that if the metadata in the store is changed after this
    consolidation, then the metadata read by :func:`open_consolidated`
    would be incorrect unless this function is called again.

    .. note:: This is an experimental feature.

    Parameters
    ----------
    store : MutableMapping or string
        Store or path to directory in file system or name of zip file.
    metadata_key : str
        Key to put the consolidated metadata under.

    Returns
    -------
    g : :class:`zarr.hierarchy.Group`
        Group instance, opened with the new consolidated metadata.

    See Also
    --------
    open_consolidated

    """
    store = normalize_store_arg(store, mode="w")

    def is_zarr_key(key):
        return (key.endswith('.zarray') or key.endswith('.zgroup') or
                key.endswith('.zattrs'))

    out = {
        'zarr_consolidated_format': 1,
        'metadata': {
            key: json_loads(store[key])
            for key in store if is_zarr_key(key)
        }
    }
    store[metadata_key] = json_dumps(out)
    return open_consolidated(store, metadata_key=metadata_key) def copy(source, dest, name=None, shallow=False, without_attrs=False, log=None,
         if_exists='raise', dry_run=False, **create_kws):
    """Copy the `source` array or group into the `dest` group.

    Parameters
    ----------
    source : group or array/dataset
        A zarr group or array, or an h5py group or dataset.
    dest : group
        A zarr or h5py group.
    name : str, optional
        Name to copy the object to.
    shallow : bool, optional
        If True, only copy immediate children of `source`.
    without_attrs : bool, optional
        Do not copy user attributes.
    log : callable, file path or file-like object, optional
        If provided, will be used to log progress information.
    if_exists : {'raise', 'replace', 'skip', 'skip_initialized'}, optional
        How to handle arrays that already exist in the destination group. If
        'raise' then a CopyError is raised on the first array already present
        in the destination group. If 'replace' then any array will be
        replaced in the destination. If 'skip' then any existing arrays will
        not be copied. If 'skip_initialized' then any existing arrays with
        all chunks initialized will not be copied (not available when copying to
        h5py).
    dry_run : bool, optional
        If True, don't actually copy anything, just log what would have
        happened.
    **create_kws
        Passed through to the create_dataset method when copying an array/dataset.

    Returns
    -------
    n_copied : int
        Number of items copied.
    n_skipped : int
        Number of items skipped.
    n_bytes_copied : int
        Number of bytes of data that were actually copied.

    Examples
    --------
    Here's an example of copying a group named 'foo' from an HDF5 file to a
    Zarr group::

        >>> import h5py
        >>> import zarr
        >>> import numpy as np
        >>> source = h5py.File('data/example.h5', mode='w')
        >>> foo = source.create_group('foo')
        >>> baz = foo.create_dataset('bar/baz', data=np.arange(100), chunks=(50,))
        >>> spam = source.create_dataset('spam', data=np.arange(100, 200), chunks=(30,))
        >>> zarr.tree(source)
        /
         ├── foo
         │   └── bar
         │       └── baz (100,) int64
         └── spam (100,) int64
        >>> dest = zarr.group()
        >>> from sys import stdout
        >>> zarr.copy(source['foo'], dest, log=stdout)
        copy /foo
        copy /foo/bar
        copy /foo/bar/baz (100,) int64
        all done: 3 copied, 0 skipped, 800 bytes copied
        (3, 0, 800)
        >>> dest.tree()  # N.B., no spam
        /
         └── foo
             └── bar
                 └── baz (100,) int64
        >>> source.close()

    The ``if_exists`` parameter provides options for how to handle pre-existing data in
    the destination. Here are some examples of these options, also using
    ``dry_run=True`` to find out what would happen without actually copying anything::

        >>> source = zarr.group()
        >>> dest = zarr.group()
        >>> baz = source.create_dataset('foo/bar/baz', data=np.arange(100))
        >>> spam = source.create_dataset('foo/spam', data=np.arange(1000))
        >>> existing_spam = dest.create_dataset('foo/spam', data=np.arange(1000))
        >>> from sys import stdout
        >>> try:
        ...     zarr.copy(source['foo'], dest, log=stdout, dry_run=True)
        ... except zarr.CopyError as e:
        ...     print(e)
        ...
        copy /foo
        copy /foo/bar
        copy /foo/bar/baz (100,) int64
        an object 'spam' already exists in destination '/foo'
        >>> zarr.copy(source['foo'], dest, log=stdout, if_exists='replace', dry_run=True)
        copy /foo
        copy /foo/bar
        copy /foo/bar/baz (100,) int64
        copy /foo/spam (1000,) int64
        dry run: 4 copied, 0 skipped
        (4, 0, 0)
        >>> zarr.copy(source['foo'], dest, log=stdout, if_exists='skip', dry_run=True)
        copy /foo
        copy /foo/bar
        copy /foo/bar/baz (100,) int64
        skip /foo/spam (1000,) int64
        dry run: 3 copied, 1 skipped
        (3, 1, 0)

    Notes
    -----
    Please note that this is an experimental feature. The behaviour of this
    function is still evolving and the default behaviour and/or parameters may change
    in future versions.

    """

    # value checks
    _check_dest_is_group(dest)

    # setup logging
    with _LogWriter(log) as log:

        # do the copying
        n_copied, n_skipped, n_bytes_copied = _copy(
            log, source, dest, name=name, root=True, shallow=shallow,
            without_attrs=without_attrs, if_exists=if_exists, dry_run=dry_run,
            **create_kws
        )

        # log a final message with a summary of what happened
        _log_copy_summary(log, dry_run, n_copied, n_skipped, n_bytes_copied)

        return n_copied, n_skipped, n_bytes_copied def copy_store(source, dest, source_path='', dest_path='', excludes=None,
               includes=None, flags=0, if_exists='raise', dry_run=False,
               log=None):
    """Copy data directly from the `source` store to the `dest` store. Use this
    function when you want to copy a group or array in the most efficient way,
    preserving all configuration and attributes. This function is more efficient
    than the copy() or copy_all() functions because it avoids de-compressing and
    re-compressing data, rather the compressed chunk data for each array are
    copied directly between stores.

    Parameters
    ----------
    source : Mapping
        Store to copy data from.
    dest : MutableMapping
        Store to copy data into.
    source_path : str, optional
        Only copy data from under this path in the source store.
    dest_path : str, optional
        Copy data into this path in the destination store.
    excludes : sequence of str, optional
        One or more regular expressions which will be matched against keys in
        the source store. Any matching key will not be copied.
    includes : sequence of str, optional
        One or more regular expressions which will be matched against keys in
        the source store and will override any excludes also matching.
    flags : int, optional
        Regular expression flags used for matching excludes and includes.
    if_exists : {'raise', 'replace', 'skip'}, optional
        How to handle keys that already exist in the destination store. If
        'raise' then a CopyError is raised on the first key already present
        in the destination store. If 'replace' then any data will be replaced in
        the destination. If 'skip' then any existing keys will not be copied.
    dry_run : bool, optional
        If True, don't actually copy anything, just log what would have
        happened.
    log : callable, file path or file-like object, optional
        If provided, will be used to log progress information.

    Returns
    -------
    n_copied : int
        Number of items copied.
    n_skipped : int
        Number of items skipped.
    n_bytes_copied : int
        Number of bytes of data that were actually copied.

    Examples
    --------

    >>> import zarr
    >>> store1 = zarr.DirectoryStore('data/example.zarr')
    >>> root = zarr.group(store1, overwrite=True)
    >>> foo = root.create_group('foo')
    >>> bar = foo.create_group('bar')
    >>> baz = bar.create_dataset('baz', shape=100, chunks=50, dtype='i8')
    >>> import numpy as np
    >>> baz[:] = np.arange(100)
    >>> root.tree()
    /
     └── foo
         └── bar
             └── baz (100,) int64
    >>> from sys import stdout
    >>> store2 = zarr.ZipStore('data/example.zip', mode='w')
    >>> zarr.copy_store(store1, store2, log=stdout)
    copy .zgroup
    copy foo/.zgroup
    copy foo/bar/.zgroup
    copy foo/bar/baz/.zarray
    copy foo/bar/baz/0
    copy foo/bar/baz/1
    all done: 6 copied, 0 skipped, 566 bytes copied
    (6, 0, 566)
    >>> new_root = zarr.group(store2)
    >>> new_root.tree()
    /
     └── foo
         └── bar
             └── baz (100,) int64
    >>> new_root['foo/bar/baz'][:]
    array([ 0,  1,  2,  ..., 97, 98, 99])
    >>> store2.close()  # zip stores need to be closed

    Notes
    -----
    Please note that this is an experimental feature. The behaviour of this
    function is still evolving and the default behaviour and/or parameters may change
    in future versions.

    """

    # normalize paths
    source_path = normalize_storage_path(source_path)
    dest_path = normalize_storage_path(dest_path)
    if source_path:
        source_path = source_path + '/'
    if dest_path:
        dest_path = dest_path + '/'

    # normalize excludes and includes
    if excludes is None:
        excludes = []
    elif isinstance(excludes, str):
        excludes = [excludes]
    if includes is None:
        includes = []
    elif isinstance(includes, str):
        includes = [includes]
    excludes = [re.compile(e, flags) for e in excludes]
    includes = [re.compile(i, flags) for i in includes]

    # check if_exists parameter
    valid_if_exists = ['raise', 'replace', 'skip']
    if if_exists not in valid_if_exists:
        raise ValueError('if_exists must be one of {!r}; found {!r}'
                         .format(valid_if_exists, if_exists))

    # setup counting variables
    n_copied = n_skipped = n_bytes_copied = 0

    # setup logging
    with _LogWriter(log) as log:

        # iterate over source keys
        for source_key in sorted(source.keys()):

            # filter to keys under source path
            if source_key.startswith(source_path):

                # process excludes and includes
                exclude = False
                for prog in excludes:
                    if prog.search(source_key):
                        exclude = True
                        break
                if exclude:
                    for prog in includes:
                        if prog.search(source_key):
                            exclude = False
                            break
                if exclude:
                    continue

                # map key to destination path
                key_suffix = source_key[len(source_path):]
                dest_key = dest_path + key_suffix

                # create a descriptive label for this operation
                descr = source_key
                if dest_key != source_key:
                    descr = descr + ' -> ' + dest_key

                # decide what to do
                do_copy = True
                if if_exists != 'replace':
                    if dest_key in dest:
                        if if_exists == 'raise':
                            raise CopyError('key {!r} exists in destination'
                                            .format(dest_key))
                        elif if_exists == 'skip':
                            do_copy = False

                # take action
                if do_copy:
                    log('copy {}'.format(descr))
                    if not dry_run:
                        data = source[source_key]
                        n_bytes_copied += buffer_size(data)
                        dest[dest_key] = data
                    n_copied += 1
                else:
                    log('skip {}'.format(descr))
                    n_skipped += 1

        # log a final message with a summary of what happened
        _log_copy_summary(log, dry_run, n_copied, n_skipped, n_bytes_copied)

    return n_copied, n_skipped, n_bytes_copied def load(store: StoreLike):
    """Load data from an array or group into memory.

    Parameters
    ----------
    store : MutableMapping or string
        Store or path to directory in file system or name of zip file.

    Returns
    -------
    out
        If the store contains an array, out will be a numpy array. If the store contains
        a group, out will be a dict-like object where keys are array names and values
        are numpy arrays.

    See Also
    --------
    save, savez

    Notes
    -----
    If loading data from a group of arrays, data will not be immediately loaded into
    memory. Rather, arrays will be loaded into memory as they are requested.

    """
    # handle polymorphic store arg
    _store = normalize_store_arg(store)
    if contains_array(_store, path=None):
        return Array(store=_store, path=None)[...]
    elif contains_group(_store, path=None):
        grp = Group(store=_store, path=None)
        return LazyLoader(grp) def open(store: StoreLike = None, mode: str = "a", **kwargs):
    """Convenience function to open a group or array using file-mode-like semantics.

    Parameters
    ----------
    store : Store or string, optional
        Store or path to directory in file system or name of zip file.
    mode : {'r', 'r+', 'a', 'w', 'w-'}, optional
        Persistence mode: 'r' means read only (must exist); 'r+' means
        read/write (must exist); 'a' means read/write (create if doesn't
        exist); 'w' means create (overwrite if exists); 'w-' means create
        (fail if exists).
    **kwargs
        Additional parameters are passed through to :func:`zarr.creation.open_array` or
        :func:`zarr.hierarchy.open_group`.

    Returns
    -------
    z : :class:`zarr.core.Array` or :class:`zarr.hierarchy.Group`
        Array or group, depending on what exists in the given store.

    See Also
    --------
    zarr.creation.open_array, zarr.hierarchy.open_group

    Examples
    --------

    Storing data in a directory 'data/example.zarr' on the local file system::

        >>> import zarr
        >>> store = 'data/example.zarr'
        >>> zw = zarr.open(store, mode='w', shape=100, dtype='i4')  # open new array
        >>> zw
        <zarr.core.Array (100,) int32>
        >>> za = zarr.open(store, mode='a')  # open existing array for reading and writing
        >>> za
        <zarr.core.Array (100,) int32>
        >>> zr = zarr.open(store, mode='r')  # open existing array read-only
        >>> zr
        <zarr.core.Array (100,) int32 read-only>
        >>> gw = zarr.open(store, mode='w')  # open new group, overwriting previous data
        >>> gw
        <zarr.hierarchy.Group '/'>
        >>> ga = zarr.open(store, mode='a')  # open existing group for reading and writing
        >>> ga
        <zarr.hierarchy.Group '/'>
        >>> gr = zarr.open(store, mode='r')  # open existing group read-only
        >>> gr
        <zarr.hierarchy.Group '/' read-only>

    """

    path = kwargs.get('path')
    # handle polymorphic store arg
    # we pass storage options explicitly, since normalize_store_arg might construct
    # a store if the input is a fsspec-compatible URL
    _store: BaseStore = normalize_store_arg(
        store, storage_options=kwargs.pop("storage_options", {}), mode=mode
    )
    path = normalize_storage_path(path)

    if mode in {'w', 'w-', 'x'}:
        if 'shape' in kwargs:
            return open_array(_store, mode=mode, **kwargs)
        else:
            return open_group(_store, mode=mode, **kwargs)

    elif mode == "a":
        if "shape" in kwargs or contains_array(_store, path):
            return open_array(_store, mode=mode, **kwargs)
        else:
            return open_group(_store, mode=mode, **kwargs)

    else:
        if contains_array(_store, path):
            return open_array(_store, mode=mode, **kwargs)
        elif contains_group(_store, path):
            return open_group(_store, mode=mode, **kwargs)
        else:
            raise PathNotFoundError(path) def open_consolidated(store: StoreLike, metadata_key=".zmetadata", mode="r+", **kwargs):
    """Open group using metadata previously consolidated into a single key.

    This is an optimised method for opening a Zarr group, where instead of
    traversing the group/array hierarchy by accessing the metadata keys at
    each level, a single key contains all of the metadata for everything.
    For remote data sources where the overhead of accessing a key is large
    compared to the time to read data.

    The group accessed must have already had its metadata consolidated into a
    single key using the function :func:`consolidate_metadata`.

    This optimised method only works in modes which do not change the
    metadata, although the data may still be written/updated.

    Parameters
    ----------
    store : MutableMapping or string
        Store or path to directory in file system or name of zip file.
    metadata_key : str
        Key to read the consolidated metadata from. The default (.zmetadata)
        corresponds to the default used by :func:`consolidate_metadata`.
    mode : {'r', 'r+'}, optional
        Persistence mode: 'r' means read only (must exist); 'r+' means
        read/write (must exist) although only writes to data are allowed,
        changes to metadata including creation of new arrays or group
        are not allowed.
    **kwargs
        Additional parameters are passed through to :func:`zarr.creation.open_array` or
        :func:`zarr.hierarchy.open_group`.

    Returns
    -------
    g : :class:`zarr.hierarchy.Group`
        Group instance, opened with the consolidated metadata.

    See Also
    --------
    consolidate_metadata

    """

    from .storage import ConsolidatedMetadataStore

    # normalize parameters
    store = normalize_store_arg(store, storage_options=kwargs.get("storage_options"), mode=mode)
    if mode not in {'r', 'r+'}:
        raise ValueError("invalid mode, expected either 'r' or 'r+'; found {!r}"
                         .format(mode))

    # setup metadata store
    meta_store = ConsolidatedMetadataStore(store, metadata_key=metadata_key)

    # pass through
    chunk_store = kwargs.pop('chunk_store', None) or store
    return open(store=meta_store, chunk_store=chunk_store, mode=mode, **kwargs) def save(store: StoreLike, *args, **kwargs):
    """Convenience function to save an array or group of arrays to the local file system.

    Parameters
    ----------
    store : MutableMapping or string
        Store or path to directory in file system or name of zip file.
    args : ndarray
        NumPy arrays with data to save.
    kwargs
        NumPy arrays with data to save.

    Examples
    --------
    Save an array to a directory on the file system (uses a :class:`DirectoryStore`)::

        >>> import zarr
        >>> import numpy as np
        >>> arr = np.arange(10000)
        >>> zarr.save('data/example.zarr', arr)
        >>> zarr.load('data/example.zarr')
        array([   0,    1,    2, ..., 9997, 9998, 9999])

    Save an array to a Zip file (uses a :class:`ZipStore`)::

        >>> zarr.save('data/example.zip', arr)
        >>> zarr.load('data/example.zip')
        array([   0,    1,    2, ..., 9997, 9998, 9999])

    Save several arrays to a directory on the file system (uses a
    :class:`DirectoryStore` and stores arrays in a group)::

        >>> import zarr
        >>> import numpy as np
        >>> a1 = np.arange(10000)
        >>> a2 = np.arange(10000, 0, -1)
        >>> zarr.save('data/example.zarr', a1, a2)
        >>> loader = zarr.load('data/example.zarr')
        >>> loader
        <LazyLoader: arr_0, arr_1>
        >>> loader['arr_0']
        array([   0,    1,    2, ..., 9997, 9998, 9999])
        >>> loader['arr_1']
        array([10000,  9999,  9998, ...,     3,     2,     1])

    Save several arrays using named keyword arguments::

        >>> zarr.save('data/example.zarr', foo=a1, bar=a2)
        >>> loader = zarr.load('data/example.zarr')
        >>> loader
        <LazyLoader: bar, foo>
        >>> loader['foo']
        array([   0,    1,    2, ..., 9997, 9998, 9999])
        >>> loader['bar']
        array([10000,  9999,  9998, ...,     3,     2,     1])

    Store several arrays in a single zip file (uses a :class:`ZipStore`)::

        >>> zarr.save('data/example.zip', foo=a1, bar=a2)
        >>> loader = zarr.load('data/example.zip')
        >>> loader
        <LazyLoader: bar, foo>
        >>> loader['foo']
        array([   0,    1,    2, ..., 9997, 9998, 9999])
        >>> loader['bar']
        array([10000,  9999,  9998, ...,     3,     2,     1])

    See Also
    --------
    save_array, save_group

    """
    if len(args) == 0 and len(kwargs) == 0:
        raise ValueError('at least one array must be provided')
    if len(args) == 1 and len(kwargs) == 0:
        save_array(store, args[0])
    else:
        save_group(store, *args, **kwargs) def save_group(store: StoreLike, *args, **kwargs):
    """Convenience function to save several NumPy arrays to the local file system, following a
    similar API to the NumPy savez()/savez_compressed() functions.

    Parameters
    ----------
    store : MutableMapping or string
        Store or path to directory in file system or name of zip file.
    args : ndarray
        NumPy arrays with data to save.
    kwargs
        NumPy arrays with data to save.

    Examples
    --------
    Save several arrays to a directory on the file system (uses a
    :class:`DirectoryStore`):

        >>> import zarr
        >>> import numpy as np
        >>> a1 = np.arange(10000)
        >>> a2 = np.arange(10000, 0, -1)
        >>> zarr.save_group('data/example.zarr', a1, a2)
        >>> loader = zarr.load('data/example.zarr')
        >>> loader
        <LazyLoader: arr_0, arr_1>
        >>> loader['arr_0']
        array([   0,    1,    2, ..., 9997, 9998, 9999])
        >>> loader['arr_1']
        array([10000,  9999,  9998, ...,     3,     2,     1])

    Save several arrays using named keyword arguments::

        >>> zarr.save_group('data/example.zarr', foo=a1, bar=a2)
        >>> loader = zarr.load('data/example.zarr')
        >>> loader
        <LazyLoader: bar, foo>
        >>> loader['foo']
        array([   0,    1,    2, ..., 9997, 9998, 9999])
        >>> loader['bar']
        array([10000,  9999,  9998, ...,     3,     2,     1])

    Store several arrays in a single zip file (uses a :class:`ZipStore`)::

        >>> zarr.save_group('data/example.zip', foo=a1, bar=a2)
        >>> loader = zarr.load('data/example.zip')
        >>> loader
        <LazyLoader: bar, foo>
        >>> loader['foo']
        array([   0,    1,    2, ..., 9997, 9998, 9999])
        >>> loader['bar']
        array([10000,  9999,  9998, ...,     3,     2,     1])

    Notes
    -----
    Default compression options will be used.

    """
    if len(args) == 0 and len(kwargs) == 0:
        raise ValueError('at least one array must be provided')
    # handle polymorphic store arg
    may_need_closing = _might_close(store)
    _store: BaseStore = normalize_store_arg(store, mode="w")
    try:
        grp = _create_group(_store, overwrite=True)
        for i, arr in enumerate(args):
            k = 'arr_{}'.format(i)
            grp.create_dataset(k, data=arr, overwrite=True)
        for k, arr in kwargs.items():
            grp.create_dataset(k, data=arr, overwrite=True)
    finally:
        if may_need_closing:
            # needed to ensure zip file records are written
            _store.close() def save_array(store: StoreLike, arr, **kwargs):
    """Convenience function to save a NumPy array to the local file system, following a
    similar API to the NumPy save() function.

    Parameters
    ----------
    store : MutableMapping or string
        Store or path to directory in file system or name of zip file.
    arr : ndarray
        NumPy array with data to save.
    kwargs
        Passed through to :func:`create`, e.g., compressor.

    Examples
    --------
    Save an array to a directory on the file system (uses a :class:`DirectoryStore`)::

        >>> import zarr
        >>> import numpy as np
        >>> arr = np.arange(10000)
        >>> zarr.save_array('data/example.zarr', arr)
        >>> zarr.load('data/example.zarr')
        array([   0,    1,    2, ..., 9997, 9998, 9999])

    Save an array to a single file (uses a :class:`ZipStore`)::

        >>> zarr.save_array('data/example.zip', arr)
        >>> zarr.load('data/example.zip')
        array([   0,    1,    2, ..., 9997, 9998, 9999])

    """
    may_need_closing = _might_close(store)
    _store: BaseStore = normalize_store_arg(store, mode="w")
    try:
        _create_array(arr, store=_store, overwrite=True, **kwargs)
    finally:
        if may_need_closing:
            # needed to ensure zip file records are written
            _store.close() def copy_all(source, dest, shallow=False, without_attrs=False, log=None,
             if_exists='raise', dry_run=False, **create_kws):
    """Copy all children of the `source` group into the `dest` group.

    Parameters
    ----------
    source : group or array/dataset
        A zarr group or array, or an h5py group or dataset.
    dest : group
        A zarr or h5py group.
    shallow : bool, optional
        If True, only copy immediate children of `source`.
    without_attrs : bool, optional
        Do not copy user attributes.
    log : callable, file path or file-like object, optional
        If provided, will be used to log progress information.
    if_exists : {'raise', 'replace', 'skip', 'skip_initialized'}, optional
        How to handle arrays that already exist in the destination group. If
        'raise' then a CopyError is raised on the first array already present
        in the destination group. If 'replace' then any array will be
        replaced in the destination. If 'skip' then any existing arrays will
        not be copied. If 'skip_initialized' then any existing arrays with
        all chunks initialized will not be copied (not available when copying to
        h5py).
    dry_run : bool, optional
        If True, don't actually copy anything, just log what would have
        happened.
    **create_kws
        Passed through to the create_dataset method when copying an
        array/dataset.

    Returns
    -------
    n_copied : int
        Number of items copied.
    n_skipped : int
        Number of items skipped.
    n_bytes_copied : int
        Number of bytes of data that were actually copied.

    Examples
    --------
    >>> import h5py
    >>> import zarr
    >>> import numpy as np
    >>> source = h5py.File('data/example.h5', mode='w')
    >>> foo = source.create_group('foo')
    >>> baz = foo.create_dataset('bar/baz', data=np.arange(100), chunks=(50,))
    >>> spam = source.create_dataset('spam', data=np.arange(100, 200), chunks=(30,))
    >>> zarr.tree(source)
    /
     ├── foo
     │   └── bar
     │       └── baz (100,) int64
     └── spam (100,) int64
    >>> dest = zarr.group()
    >>> import sys
    >>> zarr.copy_all(source, dest, log=sys.stdout)
    copy /foo
    copy /foo/bar
    copy /foo/bar/baz (100,) int64
    copy /spam (100,) int64
    all done: 4 copied, 0 skipped, 1,600 bytes copied
    (4, 0, 1600)
    >>> dest.tree()
    /
     ├── foo
     │   └── bar
     │       └── baz (100,) int64
     └── spam (100,) int64
    >>> source.close()

    Notes
    -----
    Please note that this is an experimental feature. The behaviour of this
    function is still evolving and the default behaviour and/or parameters may change
    in future versions.

    """

    # value checks
    _check_dest_is_group(dest)

    # setup counting variables
    n_copied = n_skipped = n_bytes_copied = 0

    # setup logging
    with _LogWriter(log) as log:

        for k in source.keys():
            c, s, b = _copy(
                log, source[k], dest, name=k, root=False, shallow=shallow,
                without_attrs=without_attrs, if_exists=if_exists,
                dry_run=dry_run, **create_kws)
            n_copied += c
            n_skipped += s
            n_bytes_copied += b
        dest.attrs.update(**source.attrs)

        # log a final message with a summary of what happened
        _log_copy_summary(log, dry_run, n_copied, n_skipped, n_bytes_copied)

    return n_copied, n_skipped, n_bytes_copied class Array:
    """Instantiate an array from an initialized store.

    Parameters
    ----------
    store : MutableMapping
        Array store, already initialized.
    path : string, optional
        Storage path.
    read_only : bool, optional
        True if array should be protected against modification.
    chunk_store : MutableMapping, optional
        Separate storage for chunks. If not provided, `store` will be used
        for storage of both chunks and metadata.
    synchronizer : object, optional
        Array synchronizer.
    cache_metadata : bool, optional
        If True (default), array configuration metadata will be cached for the
        lifetime of the object. If False, array metadata will be reloaded
        prior to all data access and modification operations (may incur
        overhead depending on storage and data access pattern).
    cache_attrs : bool, optional
        If True (default), user attributes will be cached for attribute read
        operations. If False, user attributes are reloaded from the store prior
        to all attribute read operations.
    partial_decompress : bool, optional
        If True and while the chunk_store is a FSStore and the compression used
        is Blosc, when getting data from the array chunks will be partially
        read and decompressed when possible.

        .. versionadded:: 2.7

    write_empty_chunks : bool, optional
        If True, all chunks will be stored regardless of their contents. If
        False (default), each chunk is compared to the array's fill value prior
        to storing. If a chunk is uniformly equal to the fill value, then that
        chunk is not be stored, and the store entry for that chunk's key is
        deleted. This setting enables sparser storage, as only chunks with
        non-fill-value data are stored, at the expense of overhead associated
        with checking the data of each chunk.

        .. versionadded:: 2.11


    Attributes
    ----------
    store
    path
    name
    read_only
    chunk_store
    shape
    chunks
    dtype
    compression
    compression_opts
    dimension_separator
    fill_value
    order
    synchronizer
    filters
    attrs
    size
    itemsize
    nbytes
    nbytes_stored
    cdata_shape
    nchunks
    nchunks_initialized
    is_view
    info
    vindex
    oindex
    write_empty_chunks

    Methods
    -------
    __getitem__
    __setitem__
    get_basic_selection
    set_basic_selection
    get_orthogonal_selection
    set_orthogonal_selection
    get_mask_selection
    set_mask_selection
    get_coordinate_selection
    set_coordinate_selection
    digest
    hexdigest
    resize
    append
    view
    astype

    """

    def __init__(
        self,
        store: BaseStore,
        path=None,
        read_only=False,
        chunk_store=None,
        synchronizer=None,
        cache_metadata=True,
        cache_attrs=True,
        partial_decompress=False,
        write_empty_chunks=False,
    ):
        # N.B., expect at this point store is fully initialized with all
        # configuration metadata fully specified and normalized

        store = BaseStore._ensure_store(store)
        chunk_store = BaseStore._ensure_store(chunk_store)

        self._store = store
        self._chunk_store = chunk_store
        self._path = normalize_storage_path(path)
        if self._path:
            self._key_prefix = self._path + '/'
        else:
            self._key_prefix = ''
        self._read_only = bool(read_only)
        self._synchronizer = synchronizer
        self._cache_metadata = cache_metadata
        self._is_view = False
        self._partial_decompress = partial_decompress
        self._write_empty_chunks = write_empty_chunks

        # initialize metadata
        self._load_metadata()

        # initialize attributes
        akey = self._key_prefix + attrs_key
        self._attrs = Attributes(store, key=akey, read_only=read_only,
                                 synchronizer=synchronizer, cache=cache_attrs)

        # initialize info reporter
        self._info_reporter = InfoReporter(self)

        # initialize indexing helpers
        self._oindex = OIndex(self)
        self._vindex = VIndex(self)

    def _load_metadata(self):
        """(Re)load metadata from store."""
        if self._synchronizer is None:
            self._load_metadata_nosync()
        else:
            mkey = self._key_prefix + array_meta_key
            with self._synchronizer[mkey]:
                self._load_metadata_nosync()

    def _load_metadata_nosync(self):
        try:
            mkey = self._key_prefix + array_meta_key
            meta_bytes = self._store[mkey]
        except KeyError:
            raise ArrayNotFoundError(self._path)
        else:

            # decode and store metadata as instance members
            meta = self._store._metadata_class.decode_array_metadata(meta_bytes)
            self._meta = meta
            self._shape = meta['shape']
            self._chunks = meta['chunks']
            self._dtype = meta['dtype']
            self._fill_value = meta['fill_value']
            self._order = meta['order']

            dimension_separator = meta.get('dimension_separator', None)
            if dimension_separator is None:
                try:
                    dimension_separator = self._store._dimension_separator
                except (AttributeError, KeyError):
                    pass

                # Fallback for any stores which do not choose a default
                if dimension_separator is None:
                    dimension_separator = "."
            self._dimension_separator = dimension_separator

            # setup compressor
            config = meta['compressor']
            if config is None:
                self._compressor = None
            else:
                self._compressor = get_codec(config)

            # setup filters
            filters = meta['filters']
            if filters:
                filters = [get_codec(config) for config in filters]
            self._filters = filters

    def _refresh_metadata(self):
        if not self._cache_metadata:
            self._load_metadata()

    def _refresh_metadata_nosync(self):
        if not self._cache_metadata and not self._is_view:
            self._load_metadata_nosync()

    def _flush_metadata_nosync(self):
        if self._is_view:
            raise PermissionError('operation not permitted for views')

        if self._compressor:
            compressor_config = self._compressor.get_config()
        else:
            compressor_config = None
        if self._filters:
            filters_config = [f.get_config() for f in self._filters]
        else:
            filters_config = None
        meta = dict(shape=self._shape, chunks=self._chunks, dtype=self._dtype,
                    compressor=compressor_config, fill_value=self._fill_value,
                    order=self._order, filters=filters_config)
        mkey = self._key_prefix + array_meta_key
        self._store[mkey] = self._store._metadata_class.encode_array_metadata(meta)

    @property
    def store(self):
        """A MutableMapping providing the underlying storage for the array."""
        return self._store

    @property
    def path(self):
        """Storage path."""
        return self._path

    @property
    def name(self):
        """Array name following h5py convention."""
        if self.path:
            # follow h5py convention: add leading slash
            name = self.path
            if name[0] != '/':
                name = '/' + name
            return name
        return None

    @property
    def basename(self):
        """Final component of name."""
        if self.name is not None:
            return self.name.split('/')[-1]
        return None

    @property
    def read_only(self):
        """A boolean, True if modification operations are not permitted."""
        return self._read_only

    @read_only.setter
    def read_only(self, value):
        self._read_only = bool(value)

    @property
    def chunk_store(self):
        """A MutableMapping providing the underlying storage for array chunks."""
        if self._chunk_store is None:
            return self._store
        else:
            return self._chunk_store

    @property
    def shape(self):
        """A tuple of integers describing the length of each dimension of
        the array."""
        # N.B., shape may change if array is resized, hence need to refresh
        # metadata
        self._refresh_metadata()
        return self._shape

    @shape.setter
    def shape(self, value):
        self.resize(value)

    @property
    def chunks(self):
        """A tuple of integers describing the length of each dimension of a
        chunk of the array."""
        return self._chunks

    @property
    def dtype(self):
        """The NumPy data type."""
        return self._dtype

    @property
    def compressor(self):
        """Primary compression codec."""
        return self._compressor

    @property
    def fill_value(self):
        """A value used for uninitialized portions of the array."""
        return self._fill_value

    @fill_value.setter
    def fill_value(self, new):
        self._fill_value = new
        self._flush_metadata_nosync()

    @property
    def order(self):
        """A string indicating the order in which bytes are arranged within
        chunks of the array."""
        return self._order

    @property
    def filters(self):
        """One or more codecs used to transform data prior to compression."""
        return self._filters

    @property
    def synchronizer(self):
        """Object used to synchronize write access to the array."""
        return self._synchronizer

    @property
    def attrs(self):
        """A MutableMapping containing user-defined attributes. Note that
        attribute values must be JSON serializable."""
        return self._attrs

    @property
    def ndim(self):
        """Number of dimensions."""
        return len(self._shape)

    @property
    def _size(self):
        return reduce(operator.mul, self._shape, 1)

    @property
    def size(self):
        """The total number of elements in the array."""
        # N.B., this property depends on shape, and shape may change if array
        # is resized, hence need to refresh metadata
        self._refresh_metadata()
        return self._size

    @property
    def itemsize(self):
        """The size in bytes of each item in the array."""
        return self.dtype.itemsize

    @property
    def _nbytes(self):
        return self._size * self.itemsize

    @property
    def nbytes(self):
        """The total number of bytes that would be required to store the
        array without compression."""
        # N.B., this property depends on shape, and shape may change if array
        # is resized, hence need to refresh metadata
        self._refresh_metadata()
        return self._nbytes

    @property
    def nbytes_stored(self):
        """The total number of stored bytes of data for the array. This
        includes storage required for configuration metadata and user
        attributes."""
        m = getsize(self._store, self._path)
        if self._chunk_store is None:
            return m
        else:
            n = getsize(self._chunk_store, self._path)
            if m < 0 or n < 0:
                return -1
            else:
                return m + n

    @property
    def _cdata_shape(self):
        if self._shape == ():
            return 1,
        else:
            return tuple(math.ceil(s / c)
                         for s, c in zip(self._shape, self._chunks))

    @property
    def cdata_shape(self):
        """A tuple of integers describing the number of chunks along each
        dimension of the array."""
        self._refresh_metadata()
        return self._cdata_shape

    @property
    def _nchunks(self):
        return reduce(operator.mul, self._cdata_shape, 1)

    @property
    def nchunks(self):
        """Total number of chunks."""
        self._refresh_metadata()
        return self._nchunks

    @property
    def nchunks_initialized(self):
        """The number of chunks that have been initialized with some data."""

        # key pattern for chunk keys
        prog = re.compile(r'\.'.join([r'\d+'] * min(1, self.ndim)))

        # count chunk keys
        return sum(1 for k in listdir(self.chunk_store, self._path) if prog.match(k))

    # backwards compatibility
    initialized = nchunks_initialized

    @property
    def is_view(self):
        """A boolean, True if this array is a view on another array."""
        return self._is_view

    @property
    def oindex(self):
        """Shortcut for orthogonal (outer) indexing, see :func:`get_orthogonal_selection` and
        :func:`set_orthogonal_selection` for documentation and examples."""
        return self._oindex

    @property
    def vindex(self):
        """Shortcut for vectorized (inner) indexing, see :func:`get_coordinate_selection`,
        :func:`set_coordinate_selection`, :func:`get_mask_selection` and
        :func:`set_mask_selection` for documentation and examples."""
        return self._vindex

    @property
    def write_empty_chunks(self) -> bool:
        """A Boolean, True if chunks composed of the array's fill value
        will be stored. If False, such chunks will not be stored.
        """
        return self._write_empty_chunks

    def __eq__(self, other):
        return (
            isinstance(other, Array) and
            self.store == other.store and
            self.read_only == other.read_only and
            self.path == other.path and
            not self._is_view
            # N.B., no need to compare other properties, should be covered by
            # store comparison
        )

    def __array__(self, *args):
        a = self[...]
        if args:
            a = a.astype(args[0])
        return a

    def islice(self, start=None, end=None):
        """
        Yield a generator for iterating over the entire or parts of the
        array. Uses a cache so chunks only have to be decompressed once.

        Parameters
        ----------
        start : int, optional
            Start index for the generator to start at. Defaults to 0.
        end : int, optional
            End index for the generator to stop at. Defaults to self.shape[0].

        Yields
        ------
        out : generator
            A generator that can be used to iterate over the requested region
            the array.

        Examples
        --------
        Setup a 1-dimensional array::

            >>> import zarr
            >>> import numpy as np
            >>> z = zarr.array(np.arange(100))

        Iterate over part of the array:
            >>> for value in z.islice(25, 30): value;
            25
            26
            27
            28
            29
        """

        if len(self.shape) == 0:
            # Same error as numpy
            raise TypeError("iteration over a 0-d array")
        if start is None:
            start = 0
        if end is None or end > self.shape[0]:
            end = self.shape[0]

        if not isinstance(start, int) or start < 0:
            raise ValueError('start must be a nonnegative integer')

        if not isinstance(end, int) or end < 0:
            raise ValueError('end must be a nonnegative integer')

        # Avoid repeatedly decompressing chunks by iterating over the chunks
        # in the first dimension.
        chunk_size = self.chunks[0]
        chunk = None
        for j in range(start, end):
            if j % chunk_size == 0:
                chunk = self[j: j + chunk_size]
            # init chunk if we start offset of chunk borders
            elif chunk is None:
                chunk_start = j - j % chunk_size
                chunk_end = chunk_start + chunk_size
                chunk = self[chunk_start:chunk_end]
            yield chunk[j % chunk_size]

    def __iter__(self):
        return self.islice()

    def __len__(self):
        if self.shape:
            return self.shape[0]
        else:
            # 0-dimensional array, same error message as numpy
            raise TypeError('len() of unsized object')

    def __getitem__(self, selection):
        """Retrieve data for an item or region of the array.

        Parameters
        ----------
        selection : tuple
            An integer index or slice or tuple of int/slice objects specifying the
            requested item or region for each dimension of the array.

        Returns
        -------
        out : ndarray
            A NumPy array containing the data for the requested region.

        Examples
        --------
        Setup a 1-dimensional array::

            >>> import zarr
            >>> import numpy as np
            >>> z = zarr.array(np.arange(100))

        Retrieve a single item::

            >>> z[5]
            5

        Retrieve a region via slicing::

            >>> z[:5]
            array([0, 1, 2, 3, 4])
            >>> z[-5:]
            array([95, 96, 97, 98, 99])
            >>> z[5:10]
            array([5, 6, 7, 8, 9])
            >>> z[5:10:2]
            array([5, 7, 9])
            >>> z[::2]
            array([ 0,  2,  4, ..., 94, 96, 98])

        Load the entire array into memory::

            >>> z[...]
            array([ 0,  1,  2, ..., 97, 98, 99])

        Setup a 2-dimensional array::

            >>> z = zarr.array(np.arange(100).reshape(10, 10))

        Retrieve an item::

            >>> z[2, 2]
            22

        Retrieve a region via slicing::

            >>> z[1:3, 1:3]
            array([[11, 12],
                   [21, 22]])
            >>> z[1:3, :]
            array([[10, 11, 12, 13, 14, 15, 16, 17, 18, 19],
                   [20, 21, 22, 23, 24, 25, 26, 27, 28, 29]])
            >>> z[:, 1:3]
            array([[ 1,  2],
                   [11, 12],
                   [21, 22],
                   [31, 32],
                   [41, 42],
                   [51, 52],
                   [61, 62],
                   [71, 72],
                   [81, 82],
                   [91, 92]])
            >>> z[0:5:2, 0:5:2]
            array([[ 0,  2,  4],
                   [20, 22, 24],
                   [40, 42, 44]])
            >>> z[::2, ::2]
            array([[ 0,  2,  4,  6,  8],
                   [20, 22, 24, 26, 28],
                   [40, 42, 44, 46, 48],
                   [60, 62, 64, 66, 68],
                   [80, 82, 84, 86, 88]])

        Load the entire array into memory::

            >>> z[...]
            array([[ 0,  1,  2,  3,  4,  5,  6,  7,  8,  9],
                   [10, 11, 12, 13, 14, 15, 16, 17, 18, 19],
                   [20, 21, 22, 23, 24, 25, 26, 27, 28, 29],
                   [30, 31, 32, 33, 34, 35, 36, 37, 38, 39],
                   [40, 41, 42, 43, 44, 45, 46, 47, 48, 49],
                   [50, 51, 52, 53, 54, 55, 56, 57, 58, 59],
                   [60, 61, 62, 63, 64, 65, 66, 67, 68, 69],
                   [70, 71, 72, 73, 74, 75, 76, 77, 78, 79],
                   [80, 81, 82, 83, 84, 85, 86, 87, 88, 89],
                   [90, 91, 92, 93, 94, 95, 96, 97, 98, 99]])

        For arrays with a structured dtype, specific fields can be retrieved, e.g.::

            >>> a = np.array([(b'aaa', 1, 4.2),
            ...               (b'bbb', 2, 8.4),
            ...               (b'ccc', 3, 12.6)],
            ...              dtype=[('foo', 'S3'), ('bar', 'i4'), ('baz', 'f8')])
            >>> z = zarr.array(a)
            >>> z['foo']
            array([b'aaa', b'bbb', b'ccc'],
                  dtype='|S3')

        Notes
        -----
        Slices with step > 1 are supported, but slices with negative step are not.

        Currently the implementation for __getitem__ is provided by
        :func:`vindex` if the indexing is pure fancy indexing (ie a
        broadcast-compatible tuple of integer array indices), or by
        :func:`set_basic_selection` otherwise.

        Effectively, this means that the following indexing modes are supported:

           - integer indexing
           - slice indexing
           - mixed slice and integer indexing
           - boolean indexing
           - fancy indexing (vectorized list of integers)

        For specific indexing options including outer indexing, see the
        methods listed under See Also.

        See Also
        --------
        get_basic_selection, set_basic_selection, get_mask_selection, set_mask_selection,
        get_coordinate_selection, set_coordinate_selection, get_orthogonal_selection,
        set_orthogonal_selection, vindex, oindex, __setitem__

        """
        fields, pure_selection = pop_fields(selection)
        if is_pure_fancy_indexing(pure_selection, self.ndim):
            result = self.vindex[selection]
        else:
            result = self.get_basic_selection(pure_selection, fields=fields)
        return result

    def get_basic_selection(self, selection=Ellipsis, out=None, fields=None):
        """Retrieve data for an item or region of the array.

        Parameters
        ----------
        selection : tuple
            A tuple specifying the requested item or region for each dimension of the
            array. May be any combination of int and/or slice for multidimensional arrays.
        out : ndarray, optional
            If given, load the selected data directly into this array.
        fields : str or sequence of str, optional
            For arrays with a structured dtype, one or more fields can be specified to
            extract data for.

        Returns
        -------
        out : ndarray
            A NumPy array containing the data for the requested region.

        Examples
        --------
        Setup a 1-dimensional array::

            >>> import zarr
            >>> import numpy as np
            >>> z = zarr.array(np.arange(100))

        Retrieve a single item::

            >>> z.get_basic_selection(5)
            5

        Retrieve a region via slicing::

            >>> z.get_basic_selection(slice(5))
            array([0, 1, 2, 3, 4])
            >>> z.get_basic_selection(slice(-5, None))
            array([95, 96, 97, 98, 99])
            >>> z.get_basic_selection(slice(5, 10))
            array([5, 6, 7, 8, 9])
            >>> z.get_basic_selection(slice(5, 10, 2))
            array([5, 7, 9])
            >>> z.get_basic_selection(slice(None, None, 2))
            array([  0,  2,  4, ..., 94, 96, 98])

        Setup a 2-dimensional array::

            >>> z = zarr.array(np.arange(100).reshape(10, 10))

        Retrieve an item::

            >>> z.get_basic_selection((2, 2))
            22

        Retrieve a region via slicing::

            >>> z.get_basic_selection((slice(1, 3), slice(1, 3)))
            array([[11, 12],
                   [21, 22]])
            >>> z.get_basic_selection((slice(1, 3), slice(None)))
            array([[10, 11, 12, 13, 14, 15, 16, 17, 18, 19],
                   [20, 21, 22, 23, 24, 25, 26, 27, 28, 29]])
            >>> z.get_basic_selection((slice(None), slice(1, 3)))
            array([[ 1,  2],
                   [11, 12],
                   [21, 22],
                   [31, 32],
                   [41, 42],
                   [51, 52],
                   [61, 62],
                   [71, 72],
                   [81, 82],
                   [91, 92]])
            >>> z.get_basic_selection((slice(0, 5, 2), slice(0, 5, 2)))
            array([[ 0,  2,  4],
                   [20, 22, 24],
                   [40, 42, 44]])
            >>> z.get_basic_selection((slice(None, None, 2), slice(None, None, 2)))
            array([[ 0,  2,  4,  6,  8],
                   [20, 22, 24, 26, 28],
                   [40, 42, 44, 46, 48],
                   [60, 62, 64, 66, 68],
                   [80, 82, 84, 86, 88]])

        For arrays with a structured dtype, specific fields can be retrieved, e.g.::

            >>> a = np.array([(b'aaa', 1, 4.2),
            ...               (b'bbb', 2, 8.4),
            ...               (b'ccc', 3, 12.6)],
            ...              dtype=[('foo', 'S3'), ('bar', 'i4'), ('baz', 'f8')])
            >>> z = zarr.array(a)
            >>> z.get_basic_selection(slice(2), fields='foo')
            array([b'aaa', b'bbb'],
                  dtype='|S3')

        Notes
        -----
        Slices with step > 1 are supported, but slices with negative step are not.

        Currently this method provides the implementation for accessing data via the
        square bracket notation (__getitem__). See :func:`__getitem__` for examples
        using the alternative notation.

        See Also
        --------
        set_basic_selection, get_mask_selection, set_mask_selection,
        get_coordinate_selection, set_coordinate_selection, get_orthogonal_selection,
        set_orthogonal_selection, vindex, oindex, __getitem__, __setitem__

        """

        # refresh metadata
        if not self._cache_metadata:
            self._load_metadata()

        # check args
        check_fields(fields, self._dtype)

        # handle zero-dimensional arrays
        if self._shape == ():
            return self._get_basic_selection_zd(selection=selection, out=out,
                                                fields=fields)
        else:
            return self._get_basic_selection_nd(selection=selection, out=out,
                                                fields=fields)

    def _get_basic_selection_zd(self, selection, out=None, fields=None):
        # special case basic selection for zero-dimensional array

        # check selection is valid
        selection = ensure_tuple(selection)
        if selection not in ((), (Ellipsis,)):
            err_too_many_indices(selection, ())

        try:
            # obtain encoded data for chunk
            ckey = self._chunk_key((0,))
            cdata = self.chunk_store[ckey]

        except KeyError:
            # chunk not initialized
            chunk = np.zeros((), dtype=self._dtype)
            if self._fill_value is not None:
                chunk.fill(self._fill_value)

        else:
            chunk = self._decode_chunk(cdata)

        # handle fields
        if fields:
            chunk = chunk[fields]

        # handle selection of the scalar value via empty tuple
        if out is None:
            out = chunk[selection]
        else:
            out[selection] = chunk[selection]

        return out

    def _get_basic_selection_nd(self, selection, out=None, fields=None):
        # implementation of basic selection for array with at least one dimension

        # setup indexer
        indexer = BasicIndexer(selection, self)

        return self._get_selection(indexer=indexer, out=out, fields=fields)

    def get_orthogonal_selection(self, selection, out=None, fields=None):
        """Retrieve data by making a selection for each dimension of the array. For
        example, if an array has 2 dimensions, allows selecting specific rows and/or
        columns. The selection for each dimension can be either an integer (indexing a
        single item), a slice, an array of integers, or a Boolean array where True
        values indicate a selection.

        Parameters
        ----------
        selection : tuple
            A selection for each dimension of the array. May be any combination of int,
            slice, integer array or Boolean array.
        out : ndarray, optional
            If given, load the selected data directly into this array.
        fields : str or sequence of str, optional
            For arrays with a structured dtype, one or more fields can be specified to
            extract data for.

        Returns
        -------
        out : ndarray
            A NumPy array containing the data for the requested selection.

        Examples
        --------
        Setup a 2-dimensional array::

            >>> import zarr
            >>> import numpy as np
            >>> z = zarr.array(np.arange(100).reshape(10, 10))

        Retrieve rows and columns via any combination of int, slice, integer array and/or
        Boolean array::

            >>> z.get_orthogonal_selection(([1, 4], slice(None)))
            array([[10, 11, 12, 13, 14, 15, 16, 17, 18, 19],
                   [40, 41, 42, 43, 44, 45, 46, 47, 48, 49]])
            >>> z.get_orthogonal_selection((slice(None), [1, 4]))
            array([[ 1,  4],
                   [11, 14],
                   [21, 24],
                   [31, 34],
                   [41, 44],
                   [51, 54],
                   [61, 64],
                   [71, 74],
                   [81, 84],
                   [91, 94]])
            >>> z.get_orthogonal_selection(([1, 4], [1, 4]))
            array([[11, 14],
                   [41, 44]])
            >>> sel = np.zeros(z.shape[0], dtype=bool)
            >>> sel[1] = True
            >>> sel[4] = True
            >>> z.get_orthogonal_selection((sel, sel))
            array([[11, 14],
                   [41, 44]])

        For convenience, the orthogonal selection functionality is also available via the
        `oindex` property, e.g.::

            >>> z.oindex[[1, 4], :]
            array([[10, 11, 12, 13, 14, 15, 16, 17, 18, 19],
                   [40, 41, 42, 43, 44, 45, 46, 47, 48, 49]])
            >>> z.oindex[:, [1, 4]]
            array([[ 1,  4],
                   [11, 14],
                   [21, 24],
                   [31, 34],
                   [41, 44],
                   [51, 54],
                   [61, 64],
                   [71, 74],
                   [81, 84],
                   [91, 94]])
            >>> z.oindex[[1, 4], [1, 4]]
            array([[11, 14],
                   [41, 44]])
            >>> sel = np.zeros(z.shape[0], dtype=bool)
            >>> sel[1] = True
            >>> sel[4] = True
            >>> z.oindex[sel, sel]
            array([[11, 14],
                   [41, 44]])

        Notes
        -----
        Orthogonal indexing is also known as outer indexing.

        Slices with step > 1 are supported, but slices with negative step are not.

        See Also
        --------
        get_basic_selection, set_basic_selection, get_mask_selection, set_mask_selection,
        get_coordinate_selection, set_coordinate_selection, set_orthogonal_selection,
        vindex, oindex, __getitem__, __setitem__

        """

        # refresh metadata
        if not self._cache_metadata:
            self._load_metadata()

        # check args
        check_fields(fields, self._dtype)

        # setup indexer
        indexer = OrthogonalIndexer(selection, self)

        return self._get_selection(indexer=indexer, out=out, fields=fields)

    def get_coordinate_selection(self, selection, out=None, fields=None):
        """Retrieve a selection of individual items, by providing the indices
        (coordinates) for each selected item.

        Parameters
        ----------
        selection : tuple
            An integer (coordinate) array for each dimension of the array.
        out : ndarray, optional
            If given, load the selected data directly into this array.
        fields : str or sequence of str, optional
            For arrays with a structured dtype, one or more fields can be specified to
            extract data for.

        Returns
        -------
        out : ndarray
            A NumPy array containing the data for the requested selection.

        Examples
        --------
        Setup a 2-dimensional array::

            >>> import zarr
            >>> import numpy as np
            >>> z = zarr.array(np.arange(100).reshape(10, 10))

        Retrieve items by specifying their coordinates::

            >>> z.get_coordinate_selection(([1, 4], [1, 4]))
            array([11, 44])

        For convenience, the coordinate selection functionality is also available via the
        `vindex` property, e.g.::

            >>> z.vindex[[1, 4], [1, 4]]
            array([11, 44])

        Notes
        -----
        Coordinate indexing is also known as point selection, and is a form of vectorized
        or inner indexing.

        Slices are not supported. Coordinate arrays must be provided for all dimensions
        of the array.

        Coordinate arrays may be multidimensional, in which case the output array will
        also be multidimensional. Coordinate arrays are broadcast against each other
        before being applied. The shape of the output will be the same as the shape of
        each coordinate array after broadcasting.

        See Also
        --------
        get_basic_selection, set_basic_selection, get_mask_selection, set_mask_selection,
        get_orthogonal_selection, set_orthogonal_selection, set_coordinate_selection,
        vindex, oindex, __getitem__, __setitem__

        """

        # refresh metadata
        if not self._cache_metadata:
            self._load_metadata()

        # check args
        check_fields(fields, self._dtype)

        # setup indexer
        indexer = CoordinateIndexer(selection, self)

        # handle output - need to flatten
        if out is not None:
            out = out.reshape(-1)

        out = self._get_selection(indexer=indexer, out=out, fields=fields)

        # restore shape
        out = out.reshape(indexer.sel_shape)

        return out

    def get_mask_selection(self, selection, out=None, fields=None):
        """Retrieve a selection of individual items, by providing a Boolean array of the
        same shape as the array against which the selection is being made, where True
        values indicate a selected item.

        Parameters
        ----------
        selection : ndarray, bool
            A Boolean array of the same shape as the array against which the selection is
            being made.
        out : ndarray, optional
            If given, load the selected data directly into this array.
        fields : str or sequence of str, optional
            For arrays with a structured dtype, one or more fields can be specified to
            extract data for.

        Returns
        -------
        out : ndarray
            A NumPy array containing the data for the requested selection.

        Examples
        --------
        Setup a 2-dimensional array::

            >>> import zarr
            >>> import numpy as np
            >>> z = zarr.array(np.arange(100).reshape(10, 10))

        Retrieve items by specifying a mask::

            >>> sel = np.zeros_like(z, dtype=bool)
            >>> sel[1, 1] = True
            >>> sel[4, 4] = True
            >>> z.get_mask_selection(sel)
            array([11, 44])

        For convenience, the mask selection functionality is also available via the
        `vindex` property, e.g.::

            >>> z.vindex[sel]
            array([11, 44])

        Notes
        -----
        Mask indexing is a form of vectorized or inner indexing, and is equivalent to
        coordinate indexing. Internally the mask array is converted to coordinate
        arrays by calling `np.nonzero`.

        See Also
        --------
        get_basic_selection, set_basic_selection, set_mask_selection,
        get_orthogonal_selection, set_orthogonal_selection, get_coordinate_selection,
        set_coordinate_selection, vindex, oindex, __getitem__, __setitem__

        """

        # refresh metadata
        if not self._cache_metadata:
            self._load_metadata()

        # check args
        check_fields(fields, self._dtype)

        # setup indexer
        indexer = MaskIndexer(selection, self)

        return self._get_selection(indexer=indexer, out=out, fields=fields)

    def _get_selection(self, indexer, out=None, fields=None):

        # We iterate over all chunks which overlap the selection and thus contain data
        # that needs to be extracted. Each chunk is processed in turn, extracting the
        # necessary data and storing into the correct location in the output array.

        # N.B., it is an important optimisation that we only visit chunks which overlap
        # the selection. This minimises the number of iterations in the main for loop.

        # check fields are sensible
        out_dtype = check_fields(fields, self._dtype)

        # determine output shape
        out_shape = indexer.shape

        # setup output array
        if out is None:
            out = np.empty(out_shape, dtype=out_dtype, order=self._order)
        else:
            check_array_shape('out', out, out_shape)

        # iterate over chunks
        if not hasattr(self.chunk_store, "getitems") or \
           any(map(lambda x: x == 0, self.shape)):
            # sequentially get one key at a time from storage
            for chunk_coords, chunk_selection, out_selection in indexer:

                # load chunk selection into output array
                self._chunk_getitem(chunk_coords, chunk_selection, out, out_selection,
                                    drop_axes=indexer.drop_axes, fields=fields)
        else:
            # allow storage to get multiple items at once
            lchunk_coords, lchunk_selection, lout_selection = zip(*indexer)
            self._chunk_getitems(lchunk_coords, lchunk_selection, out, lout_selection,
                                 drop_axes=indexer.drop_axes, fields=fields)

        if out.shape:
            return out
        else:
            return out[()]

    def __setitem__(self, selection, value):
        """Modify data for an item or region of the array.

        Parameters
        ----------
        selection : tuple
            An integer index or slice or tuple of int/slice specifying the requested
            region for each dimension of the array.
        value : scalar or array-like
            Value to be stored into the array.

        Examples
        --------
        Setup a 1-dimensional array::

            >>> import zarr
            >>> z = zarr.zeros(100, dtype=int)

        Set all array elements to the same scalar value::

            >>> z[...] = 42
            >>> z[...]
            array([42, 42, 42, ..., 42, 42, 42])

        Set a portion of the array::

            >>> z[:10] = np.arange(10)
            >>> z[-10:] = np.arange(10)[::-1]
            >>> z[...]
            array([ 0, 1, 2, ..., 2, 1, 0])

        Setup a 2-dimensional array::

            >>> z = zarr.zeros((5, 5), dtype=int)

        Set all array elements to the same scalar value::

            >>> z[...] = 42

        Set a portion of the array::

            >>> z[0, :] = np.arange(z.shape[1])
            >>> z[:, 0] = np.arange(z.shape[0])
            >>> z[...]
            array([[ 0,  1,  2,  3,  4],
                   [ 1, 42, 42, 42, 42],
                   [ 2, 42, 42, 42, 42],
                   [ 3, 42, 42, 42, 42],
                   [ 4, 42, 42, 42, 42]])

        For arrays with a structured dtype, specific fields can be modified, e.g.::

            >>> a = np.array([(b'aaa', 1, 4.2),
            ...               (b'bbb', 2, 8.4),
            ...               (b'ccc', 3, 12.6)],
            ...              dtype=[('foo', 'S3'), ('bar', 'i4'), ('baz', 'f8')])
            >>> z = zarr.array(a)
            >>> z['foo'] = b'zzz'
            >>> z[...]
            array([(b'zzz', 1,   4.2), (b'zzz', 2,   8.4), (b'zzz', 3,  12.6)],
                  dtype=[('foo', 'S3'), ('bar', '<i4'), ('baz', '<f8')])

        Notes
        -----
        Slices with step > 1 are supported, but slices with negative step are not.

        Currently the implementation for __setitem__ is provided by
        :func:`vindex` if the indexing is pure fancy indexing (ie a
        broadcast-compatible tuple of integer array indices), or by
        :func:`set_basic_selection` otherwise.

        Effectively, this means that the following indexing modes are supported:

           - integer indexing
           - slice indexing
           - mixed slice and integer indexing
           - boolean indexing
           - fancy indexing (vectorized list of integers)

        For specific indexing options including outer indexing, see the
        methods listed under See Also.

        See Also
        --------
        get_basic_selection, set_basic_selection, get_mask_selection, set_mask_selection,
        get_coordinate_selection, set_coordinate_selection, get_orthogonal_selection,
        set_orthogonal_selection, vindex, oindex, __getitem__

        """
        fields, pure_selection = pop_fields(selection)
        if is_pure_fancy_indexing(pure_selection, self.ndim):
            self.vindex[selection] = value
        else:
            self.set_basic_selection(pure_selection, value, fields=fields)

    def set_basic_selection(self, selection, value, fields=None):
        """Modify data for an item or region of the array.

        Parameters
        ----------
        selection : tuple
            An integer index or slice or tuple of int/slice specifying the requested
            region for each dimension of the array.
        value : scalar or array-like
            Value to be stored into the array.
        fields : str or sequence of str, optional
            For arrays with a structured dtype, one or more fields can be specified to set
            data for.

        Examples
        --------
        Setup a 1-dimensional array::

            >>> import zarr
            >>> import numpy as np
            >>> z = zarr.zeros(100, dtype=int)

        Set all array elements to the same scalar value::

            >>> z.set_basic_selection(..., 42)
            >>> z[...]
            array([42, 42, 42, ..., 42, 42, 42])

        Set a portion of the array::

            >>> z.set_basic_selection(slice(10), np.arange(10))
            >>> z.set_basic_selection(slice(-10, None), np.arange(10)[::-1])
            >>> z[...]
            array([ 0, 1, 2, ..., 2, 1, 0])

        Setup a 2-dimensional array::

            >>> z = zarr.zeros((5, 5), dtype=int)

        Set all array elements to the same scalar value::

            >>> z.set_basic_selection(..., 42)

        Set a portion of the array::

            >>> z.set_basic_selection((0, slice(None)), np.arange(z.shape[1]))
            >>> z.set_basic_selection((slice(None), 0), np.arange(z.shape[0]))
            >>> z[...]
            array([[ 0,  1,  2,  3,  4],
                   [ 1, 42, 42, 42, 42],
                   [ 2, 42, 42, 42, 42],
                   [ 3, 42, 42, 42, 42],
                   [ 4, 42, 42, 42, 42]])

        For arrays with a structured dtype, the `fields` parameter can be used to set
        data for a specific field, e.g.::

            >>> a = np.array([(b'aaa', 1, 4.2),
            ...               (b'bbb', 2, 8.4),
            ...               (b'ccc', 3, 12.6)],
            ...              dtype=[('foo', 'S3'), ('bar', 'i4'), ('baz', 'f8')])
            >>> z = zarr.array(a)
            >>> z.set_basic_selection(slice(0, 2), b'zzz', fields='foo')
            >>> z[:]
            array([(b'zzz', 1,   4.2), (b'zzz', 2,   8.4), (b'ccc', 3,  12.6)],
                  dtype=[('foo', 'S3'), ('bar', '<i4'), ('baz', '<f8')])

        Notes
        -----
        This method provides the underlying implementation for modifying data via square
        bracket notation, see :func:`__setitem__` for equivalent examples using the
        alternative notation.

        See Also
        --------
        get_basic_selection, get_mask_selection, set_mask_selection,
        get_coordinate_selection, set_coordinate_selection, get_orthogonal_selection,
        set_orthogonal_selection, vindex, oindex, __getitem__, __setitem__

        """

        # guard conditions
        if self._read_only:
            raise ReadOnlyError()

        # refresh metadata
        if not self._cache_metadata:
            self._load_metadata_nosync()

        # handle zero-dimensional arrays
        if self._shape == ():
            return self._set_basic_selection_zd(selection, value, fields=fields)
        else:
            return self._set_basic_selection_nd(selection, value, fields=fields)

    def set_orthogonal_selection(self, selection, value, fields=None):
        """Modify data via a selection for each dimension of the array.

        Parameters
        ----------
        selection : tuple
            A selection for each dimension of the array. May be any combination of int,
            slice, integer array or Boolean array.
        value : scalar or array-like
            Value to be stored into the array.
        fields : str or sequence of str, optional
            For arrays with a structured dtype, one or more fields can be specified to set
            data for.

        Examples
        --------
        Setup a 2-dimensional array::

            >>> import zarr
            >>> import numpy as np
            >>> z = zarr.zeros((5, 5), dtype=int)

        Set data for a selection of rows::

            >>> z.set_orthogonal_selection(([1, 4], slice(None)), 1)
            >>> z[...]
            array([[0, 0, 0, 0, 0],
                   [1, 1, 1, 1, 1],
                   [0, 0, 0, 0, 0],
                   [0, 0, 0, 0, 0],
                   [1, 1, 1, 1, 1]])

        Set data for a selection of columns::

            >>> z.set_orthogonal_selection((slice(None), [1, 4]), 2)
            >>> z[...]
            array([[0, 2, 0, 0, 2],
                   [1, 2, 1, 1, 2],
                   [0, 2, 0, 0, 2],
                   [0, 2, 0, 0, 2],
                   [1, 2, 1, 1, 2]])

        Set data for a selection of rows and columns::

            >>> z.set_orthogonal_selection(([1, 4], [1, 4]), 3)
            >>> z[...]
            array([[0, 2, 0, 0, 2],
                   [1, 3, 1, 1, 3],
                   [0, 2, 0, 0, 2],
                   [0, 2, 0, 0, 2],
                   [1, 3, 1, 1, 3]])

        For convenience, this functionality is also available via the `oindex` property.
        E.g.::

            >>> z.oindex[[1, 4], [1, 4]] = 4
            >>> z[...]
            array([[0, 2, 0, 0, 2],
                   [1, 4, 1, 1, 4],
                   [0, 2, 0, 0, 2],
                   [0, 2, 0, 0, 2],
                   [1, 4, 1, 1, 4]])

        Notes
        -----
        Orthogonal indexing is also known as outer indexing.

        Slices with step > 1 are supported, but slices with negative step are not.

        See Also
        --------
        get_basic_selection, set_basic_selection, get_mask_selection, set_mask_selection,
        get_coordinate_selection, set_coordinate_selection, get_orthogonal_selection,
        vindex, oindex, __getitem__, __setitem__

        """

        # guard conditions
        if self._read_only:
            raise ReadOnlyError()

        # refresh metadata
        if not self._cache_metadata:
            self._load_metadata_nosync()

        # setup indexer
        indexer = OrthogonalIndexer(selection, self)

        self._set_selection(indexer, value, fields=fields)

    def set_coordinate_selection(self, selection, value, fields=None):
        """Modify a selection of individual items, by providing the indices (coordinates)
        for each item to be modified.

        Parameters
        ----------
        selection : tuple
            An integer (coordinate) array for each dimension of the array.
        value : scalar or array-like
            Value to be stored into the array.
        fields : str or sequence of str, optional
            For arrays with a structured dtype, one or more fields can be specified to set
            data for.

        Examples
        --------
        Setup a 2-dimensional array::

            >>> import zarr
            >>> import numpy as np
            >>> z = zarr.zeros((5, 5), dtype=int)

        Set data for a selection of items::

            >>> z.set_coordinate_selection(([1, 4], [1, 4]), 1)
            >>> z[...]
            array([[0, 0, 0, 0, 0],
                   [0, 1, 0, 0, 0],
                   [0, 0, 0, 0, 0],
                   [0, 0, 0, 0, 0],
                   [0, 0, 0, 0, 1]])

        For convenience, this functionality is also available via the `vindex` property.
        E.g.::

            >>> z.vindex[[1, 4], [1, 4]] = 2
            >>> z[...]
            array([[0, 0, 0, 0, 0],
                   [0, 2, 0, 0, 0],
                   [0, 0, 0, 0, 0],
                   [0, 0, 0, 0, 0],
                   [0, 0, 0, 0, 2]])

        Notes
        -----
        Coordinate indexing is also known as point selection, and is a form of vectorized
        or inner indexing.

        Slices are not supported. Coordinate arrays must be provided for all dimensions
        of the array.

        See Also
        --------
        get_basic_selection, set_basic_selection, get_mask_selection, set_mask_selection,
        get_orthogonal_selection, set_orthogonal_selection, get_coordinate_selection,
        vindex, oindex, __getitem__, __setitem__

        """

        # guard conditions
        if self._read_only:
            raise ReadOnlyError()

        # refresh metadata
        if not self._cache_metadata:
            self._load_metadata_nosync()

        # setup indexer
        indexer = CoordinateIndexer(selection, self)

        # handle value - need to flatten
        if not is_scalar(value, self._dtype):
            value = np.asanyarray(value)
        if hasattr(value, 'shape') and len(value.shape) > 1:
            value = value.reshape(-1)

        self._set_selection(indexer, value, fields=fields)

    def set_mask_selection(self, selection, value, fields=None):
        """Modify a selection of individual items, by providing a Boolean array of the
        same shape as the array against which the selection is being made, where True
        values indicate a selected item.

        Parameters
        ----------
        selection : ndarray, bool
            A Boolean array of the same shape as the array against which the selection is
            being made.
        value : scalar or array-like
            Value to be stored into the array.
        fields : str or sequence of str, optional
            For arrays with a structured dtype, one or more fields can be specified to set
            data for.

        Examples
        --------
        Setup a 2-dimensional array::

            >>> import zarr
            >>> import numpy as np
            >>> z = zarr.zeros((5, 5), dtype=int)

        Set data for a selection of items::

            >>> sel = np.zeros_like(z, dtype=bool)
            >>> sel[1, 1] = True
            >>> sel[4, 4] = True
            >>> z.set_mask_selection(sel, 1)
            >>> z[...]
            array([[0, 0, 0, 0, 0],
                   [0, 1, 0, 0, 0],
                   [0, 0, 0, 0, 0],
                   [0, 0, 0, 0, 0],
                   [0, 0, 0, 0, 1]])

        For convenience, this functionality is also available via the `vindex` property.
        E.g.::

            >>> z.vindex[sel] = 2
            >>> z[...]
            array([[0, 0, 0, 0, 0],
                   [0, 2, 0, 0, 0],
                   [0, 0, 0, 0, 0],
                   [0, 0, 0, 0, 0],
                   [0, 0, 0, 0, 2]])

        Notes
        -----
        Mask indexing is a form of vectorized or inner indexing, and is equivalent to
        coordinate indexing. Internally the mask array is converted to coordinate
        arrays by calling `np.nonzero`.

        See Also
        --------
        get_basic_selection, set_basic_selection, get_mask_selection,
        get_orthogonal_selection, set_orthogonal_selection, get_coordinate_selection,
        set_coordinate_selection, vindex, oindex, __getitem__, __setitem__

        """

        # guard conditions
        if self._read_only:
            raise ReadOnlyError()

        # refresh metadata
        if not self._cache_metadata:
            self._load_metadata_nosync()

        # setup indexer
        indexer = MaskIndexer(selection, self)

        self._set_selection(indexer, value, fields=fields)

    def _set_basic_selection_zd(self, selection, value, fields=None):
        # special case __setitem__ for zero-dimensional array

        # check selection is valid
        selection = ensure_tuple(selection)
        if selection not in ((), (Ellipsis,)):
            err_too_many_indices(selection, self._shape)

        # check fields
        check_fields(fields, self._dtype)
        fields = check_no_multi_fields(fields)

        # obtain key for chunk
        ckey = self._chunk_key((0,))

        # setup chunk
        try:
            # obtain compressed data for chunk
            cdata = self.chunk_store[ckey]

        except KeyError:
            # chunk not initialized
            chunk = np.zeros((), dtype=self._dtype)
            if self._fill_value is not None:
                chunk.fill(self._fill_value)

        else:
            # decode chunk
            chunk = self._decode_chunk(cdata).copy()

        # set value
        if fields:
            chunk[fields][selection] = value
        else:
            chunk[selection] = value

        # remove chunk if write_empty_chunks is false and it only contains the fill value
        if (not self.write_empty_chunks) and all_equal(self.fill_value, chunk):
            try:
                del self.chunk_store[ckey]
                return
            except Exception:  # pragma: no cover
                # deleting failed, fallback to overwriting
                pass
        else:
            # encode and store
            cdata = self._encode_chunk(chunk)
            self.chunk_store[ckey] = cdata

    def _set_basic_selection_nd(self, selection, value, fields=None):
        # implementation of __setitem__ for array with at least one dimension

        # setup indexer
        indexer = BasicIndexer(selection, self)

        self._set_selection(indexer, value, fields=fields)

    def _set_selection(self, indexer, value, fields=None):

        # We iterate over all chunks which overlap the selection and thus contain data
        # that needs to be replaced. Each chunk is processed in turn, extracting the
        # necessary data from the value array and storing into the chunk array.

        # N.B., it is an important optimisation that we only visit chunks which overlap
        # the selection. This minimises the number of iterations in the main for loop.

        # check fields are sensible
        check_fields(fields, self._dtype)
        fields = check_no_multi_fields(fields)

        # determine indices of chunks overlapping the selection
        sel_shape = indexer.shape

        # check value shape
        if sel_shape == ():
            # setting a single item
            pass
        elif is_scalar(value, self._dtype):
            # setting a scalar value
            pass
        else:
            if not hasattr(value, 'shape'):
                value = np.asanyarray(value)
            check_array_shape('value', value, sel_shape)

        # iterate over chunks in range
        if not hasattr(self.store, "setitems") or self._synchronizer is not None \
           or any(map(lambda x: x == 0, self.shape)):
            # iterative approach
            for chunk_coords, chunk_selection, out_selection in indexer:

                # extract data to store
                if sel_shape == ():
                    chunk_value = value
                elif is_scalar(value, self._dtype):
                    chunk_value = value
                else:
                    chunk_value = value[out_selection]
                    # handle missing singleton dimensions
                    if indexer.drop_axes:
                        item = [slice(None)] * self.ndim
                        for a in indexer.drop_axes:
                            item[a] = np.newaxis
                        item = tuple(item)
                        chunk_value = chunk_value[item]

                # put data
                self._chunk_setitem(chunk_coords, chunk_selection, chunk_value, fields=fields)
        else:
            lchunk_coords, lchunk_selection, lout_selection = zip(*indexer)
            chunk_values = []
            for out_selection in lout_selection:
                if sel_shape == ():
                    chunk_values.append(value)
                elif is_scalar(value, self._dtype):
                    chunk_values.append(value)
                else:
                    cv = value[out_selection]
                    # handle missing singleton dimensions
                    if indexer.drop_axes:  # pragma: no cover
                        item = [slice(None)] * self.ndim
                        for a in indexer.drop_axes:
                            item[a] = np.newaxis
                        item = tuple(item)
                        cv = chunk_value[item]
                    chunk_values.append(cv)

            self._chunk_setitems(lchunk_coords, lchunk_selection, chunk_values,
                                 fields=fields)

    def _process_chunk(
        self,
        out,
        cdata,
        chunk_selection,
        drop_axes,
        out_is_ndarray,
        fields,
        out_selection,
        partial_read_decode=False,
    ):
        """Take binary data from storage and fill output array"""
        if (out_is_ndarray and
                not fields and
                is_contiguous_selection(out_selection) and
                is_total_slice(chunk_selection, self._chunks) and
                not self._filters and
                self._dtype != object):

            dest = out[out_selection]
            write_direct = (
                dest.flags.writeable and
                (
                    (self._order == 'C' and dest.flags.c_contiguous) or
                    (self._order == 'F' and dest.flags.f_contiguous)
                )
            )

            if write_direct:

                # optimization: we want the whole chunk, and the destination is
                # contiguous, so we can decompress directly from the chunk
                # into the destination array
                if self._compressor:
                    if isinstance(cdata, PartialReadBuffer):
                        cdata = cdata.read_full()
                    self._compressor.decode(cdata, dest)
                else:
                    chunk = ensure_ndarray(cdata).view(self._dtype)
                    chunk = chunk.reshape(self._chunks, order=self._order)
                    np.copyto(dest, chunk)
                return

        # decode chunk
        try:
            if partial_read_decode:
                cdata.prepare_chunk()
                # size of chunk
                tmp = np.empty(self._chunks, dtype=self.dtype)
                index_selection = PartialChunkIterator(chunk_selection, self.chunks)
                for start, nitems, partial_out_selection in index_selection:
                    expected_shape = [
                        len(
                            range(*partial_out_selection[i].indices(self.chunks[0] + 1))
                        )
                        if i < len(partial_out_selection)
                        else dim
                        for i, dim in enumerate(self.chunks)
                    ]
                    cdata.read_part(start, nitems)
                    chunk_partial = self._decode_chunk(
                        cdata.buff,
                        start=start,
                        nitems=nitems,
                        expected_shape=expected_shape,
                    )
                    tmp[partial_out_selection] = chunk_partial
                out[out_selection] = tmp[chunk_selection]
                return
        except ArrayIndexError:
            cdata = cdata.read_full()
        chunk = self._decode_chunk(cdata)

        # select data from chunk
        if fields:
            chunk = chunk[fields]
        tmp = chunk[chunk_selection]
        if drop_axes:
            tmp = np.squeeze(tmp, axis=drop_axes)

        # store selected data in output
        out[out_selection] = tmp

    def _chunk_getitem(self, chunk_coords, chunk_selection, out, out_selection,
                       drop_axes=None, fields=None):
        """Obtain part or whole of a chunk.

        Parameters
        ----------
        chunk_coords : tuple of ints
            Indices of the chunk.
        chunk_selection : selection
            Location of region within the chunk to extract.
        out : ndarray
            Array to store result in.
        out_selection : selection
            Location of region within output array to store results in.
        drop_axes : tuple of ints
            Axes to squeeze out of the chunk.
        fields
            TODO

        """
        out_is_ndarray = True
        try:
            out = ensure_ndarray(out)
        except TypeError:
            out_is_ndarray = False

        assert len(chunk_coords) == len(self._cdata_shape)

        # obtain key for chunk
        ckey = self._chunk_key(chunk_coords)

        try:
            # obtain compressed data for chunk
            cdata = self.chunk_store[ckey]

        except KeyError:
            # chunk not initialized
            if self._fill_value is not None:
                if fields:
                    fill_value = self._fill_value[fields]
                else:
                    fill_value = self._fill_value
                out[out_selection] = fill_value

        else:
            self._process_chunk(out, cdata, chunk_selection, drop_axes,
                                out_is_ndarray, fields, out_selection)

    def _chunk_getitems(self, lchunk_coords, lchunk_selection, out, lout_selection,
                        drop_axes=None, fields=None):
        """As _chunk_getitem, but for lists of chunks

        This gets called where the storage supports ``getitems``, so that
        it can decide how to fetch the keys, allowing concurrency.
        """
        out_is_ndarray = True
        try:
            out = ensure_ndarray(out)
        except TypeError:  # pragma: no cover
            out_is_ndarray = False

        ckeys = [self._chunk_key(ch) for ch in lchunk_coords]
        if (
            self._partial_decompress
            and self._compressor
            and self._compressor.codec_id == "blosc"
            and hasattr(self._compressor, "decode_partial")
            and not fields
            and self.dtype != object
            and hasattr(self.chunk_store, "getitems")
        ):
            partial_read_decode = True
            cdatas = {
                ckey: PartialReadBuffer(ckey, self.chunk_store)
                for ckey in ckeys
                if ckey in self.chunk_store
            }
        else:
            partial_read_decode = False
            cdatas = self.chunk_store.getitems(ckeys, on_error="omit")
        for ckey, chunk_select, out_select in zip(ckeys, lchunk_selection, lout_selection):
            if ckey in cdatas:
                self._process_chunk(
                    out,
                    cdatas[ckey],
                    chunk_select,
                    drop_axes,
                    out_is_ndarray,
                    fields,
                    out_select,
                    partial_read_decode=partial_read_decode,
                )
            else:
                # check exception type
                if self._fill_value is not None:
                    if fields:
                        fill_value = self._fill_value[fields]
                    else:
                        fill_value = self._fill_value
                    out[out_select] = fill_value

    def _chunk_setitems(self, lchunk_coords, lchunk_selection, values, fields=None):
        ckeys = map(self._chunk_key, lchunk_coords)
        cdatas = {key: self._process_for_setitem(key, sel, val, fields=fields)
                  for key, sel, val in zip(ckeys, lchunk_selection, values)}
        to_store = {}
        if not self.write_empty_chunks:
            empty_chunks = {k: v for k, v in cdatas.items() if all_equal(self.fill_value, v)}
            self._chunk_delitems(empty_chunks.keys())
            nonempty_keys = cdatas.keys() - empty_chunks.keys()
            to_store = {k: self._encode_chunk(cdatas[k]) for k in nonempty_keys}
        else:
            to_store = {k: self._encode_chunk(v) for k, v in cdatas.items()}
        self.chunk_store.setitems(to_store)

    def _chunk_delitems(self, ckeys):
        if hasattr(self.store, "delitems"):
            self.store.delitems(ckeys)
        else:  # pragma: no cover
            # exempting this branch from coverage as there are no extant stores
            # that will trigger this condition, but it's possible that they
            # will be developed in the future.
            tuple(map(self._chunk_delitem, ckeys))

    def _chunk_delitem(self, ckey):
        """
        Attempt to delete the value associated with ckey.
        """
        try:
            del self.chunk_store[ckey]
        except KeyError:
            pass

    def _chunk_setitem(self, chunk_coords, chunk_selection, value, fields=None):
        """Replace part or whole of a chunk.

        Parameters
        ----------
        chunk_coords : tuple of ints
            Indices of the chunk.
        chunk_selection : tuple of slices
            Location of region within the chunk.
        value : scalar or ndarray
            Value to set.

        """

        if self._synchronizer is None:
            # no synchronization
            lock = nolock
        else:
            # synchronize on the chunk
            ckey = self._chunk_key(chunk_coords)
            lock = self._synchronizer[ckey]

        with lock:
            self._chunk_setitem_nosync(chunk_coords, chunk_selection, value,
                                       fields=fields)

    def _chunk_setitem_nosync(self, chunk_coords, chunk_selection, value, fields=None):
        ckey = self._chunk_key(chunk_coords)
        cdata = self._process_for_setitem(ckey, chunk_selection, value, fields=fields)

        # attempt to delete chunk if it only contains the fill value
        if (not self.write_empty_chunks) and all_equal(self.fill_value, cdata):
            self._chunk_delitem(ckey)
        else:
            self.chunk_store[ckey] = self._encode_chunk(cdata)

    def _process_for_setitem(self, ckey, chunk_selection, value, fields=None):
        if is_total_slice(chunk_selection, self._chunks) and not fields:
            # totally replace chunk

            # optimization: we are completely replacing the chunk, so no need
            # to access the existing chunk data

            if is_scalar(value, self._dtype):

                # setup array filled with value
                chunk = np.empty(self._chunks, dtype=self._dtype, order=self._order)
                chunk.fill(value)

            else:

                # ensure array is contiguous
                chunk = value.astype(self._dtype, order=self._order, copy=False)

        else:
            # partially replace the contents of this chunk

            try:

                # obtain compressed data for chunk
                cdata = self.chunk_store[ckey]

            except KeyError:

                # chunk not initialized
                if self._fill_value is not None:
                    chunk = np.empty(self._chunks, dtype=self._dtype, order=self._order)
                    chunk.fill(self._fill_value)
                elif self._dtype == object:
                    chunk = np.empty(self._chunks, dtype=self._dtype, order=self._order)
                else:
                    # N.B., use zeros here so any region beyond the array has consistent
                    # and compressible data
                    chunk = np.zeros(self._chunks, dtype=self._dtype, order=self._order)

            else:

                # decode chunk
                chunk = self._decode_chunk(cdata)
                if not chunk.flags.writeable:
                    chunk = chunk.copy(order='K')

            # modify
            if fields:
                # N.B., currently multi-field assignment is not supported in numpy, so
                # this only works for a single field
                chunk[fields][chunk_selection] = value
            else:
                chunk[chunk_selection] = value

        return chunk

    def _chunk_key(self, chunk_coords):
        return self._key_prefix + self._dimension_separator.join(map(str, chunk_coords))

    def _decode_chunk(self, cdata, start=None, nitems=None, expected_shape=None):
        # decompress
        if self._compressor:
            # only decode requested items
            if (
                all(x is not None for x in [start, nitems])
                and self._compressor.codec_id == "blosc"
            ) and hasattr(self._compressor, "decode_partial"):
                chunk = self._compressor.decode_partial(cdata, start, nitems)
            else:
                chunk = self._compressor.decode(cdata)
        else:
            chunk = cdata

        # apply filters
        if self._filters:
            for f in reversed(self._filters):
                chunk = f.decode(chunk)

        # view as numpy array with correct dtype
        chunk = ensure_ndarray(chunk)
        # special case object dtype, because incorrect handling can lead to
        # segfaults and other bad things happening
        if self._dtype != object:
            chunk = chunk.view(self._dtype)
        elif chunk.dtype != object:
            # If we end up here, someone must have hacked around with the filters.
            # We cannot deal with object arrays unless there is an object
            # codec in the filter chain, i.e., a filter that converts from object
            # array to something else during encoding, and converts back to object
            # array during decoding.
            raise RuntimeError('cannot read object array without object codec')

        # ensure correct chunk shape
        chunk = chunk.reshape(-1, order='A')
        chunk = chunk.reshape(expected_shape or self._chunks, order=self._order)

        return chunk

    def _encode_chunk(self, chunk):

        # apply filters
        if self._filters:
            for f in self._filters:
                chunk = f.encode(chunk)

        # check object encoding
        if ensure_ndarray(chunk).dtype == object:
            raise RuntimeError('cannot write object array without object codec')

        # compress
        if self._compressor:
            cdata = self._compressor.encode(chunk)
        else:
            cdata = chunk

        # ensure in-memory data is immutable and easy to compare
        if isinstance(self.chunk_store, MutableMapping):
            cdata = ensure_bytes(cdata)

        return cdata

    def __repr__(self):
        t = type(self)
        r = '<{}.{}'.format(t.__module__, t.__name__)
        if self.name:
            r += ' %r' % self.name
        r += ' %s' % str(self.shape)
        r += ' %s' % self.dtype
        if self._read_only:
            r += ' read-only'
        r += '>'
        return r

    @property
    def info(self):
        """Report some diagnostic information about the array.

        Examples
        --------
        >>> import zarr
        >>> z = zarr.zeros(1000000, chunks=100000, dtype='i4')
        >>> z.info
        Type               : zarr.core.Array
        Data type          : int32
        Shape              : (1000000,)
        Chunk shape        : (100000,)
        Order              : C
        Read-only          : False
        Compressor         : Blosc(cname='lz4', clevel=5, shuffle=SHUFFLE, blocksize=0)
        Store type         : zarr.storage.KVStore
        No. bytes          : 4000000 (3.8M)
        No. bytes stored   : 320
        Storage ratio      : 12500.0
        Chunks initialized : 0/10

        """
        return self._info_reporter

    def info_items(self):
        return self._synchronized_op(self._info_items_nosync)

    def _info_items_nosync(self):

        def typestr(o):
            return '{}.{}'.format(type(o).__module__, type(o).__name__)

        def bytestr(n):
            if n > 2**10:
                return '{} ({})'.format(n, human_readable_size(n))
            else:
                return str(n)

        items = []

        # basic info
        if self.name is not None:
            items += [('Name', self.name)]
        items += [
            ('Type', typestr(self)),
            ('Data type', '%s' % self.dtype),
            ('Shape', str(self.shape)),
            ('Chunk shape', str(self.chunks)),
            ('Order', self.order),
            ('Read-only', str(self.read_only)),
        ]

        # filters
        if self.filters:
            for i, f in enumerate(self.filters):
                items += [('Filter [%s]' % i, repr(f))]

        # compressor
        items += [('Compressor', repr(self.compressor))]

        # synchronizer
        if self._synchronizer is not None:
            items += [('Synchronizer type', typestr(self._synchronizer))]

        # storage info
        items += [('Store type', typestr(self._store))]
        if self._chunk_store is not None:
            items += [('Chunk store type', typestr(self._chunk_store))]
        items += [('No. bytes', bytestr(self.nbytes))]
        if self.nbytes_stored > 0:
            items += [
                ('No. bytes stored', bytestr(self.nbytes_stored)),
                ('Storage ratio', '%.1f' % (self.nbytes / self.nbytes_stored)),
            ]
        items += [
            ('Chunks initialized', '{}/{}'.format(self.nchunks_initialized, self.nchunks))
        ]

        return items

    def digest(self, hashname="sha1"):
        """
        Compute a checksum for the data. Default uses sha1 for speed.

        Examples
        --------
        >>> import binascii
        >>> import zarr
        >>> z = zarr.empty(shape=(10000, 10000), chunks=(1000, 1000))
        >>> binascii.hexlify(z.digest())
        b'041f90bc7a571452af4f850a8ca2c6cddfa8a1ac'
        >>> z = zarr.zeros(shape=(10000, 10000), chunks=(1000, 1000))
        >>> binascii.hexlify(z.digest())
        b'7162d416d26a68063b66ed1f30e0a866e4abed60'
        >>> z = zarr.zeros(shape=(10000, 10000), dtype="u1", chunks=(1000, 1000))
        >>> binascii.hexlify(z.digest())
        b'cb387af37410ae5a3222e893cf3373e4e4f22816'
        """

        h = hashlib.new(hashname)

        for i in itertools.product(*[range(s) for s in self.cdata_shape]):
            h.update(self.chunk_store.get(self._chunk_key(i), b""))

        h.update(self.store.get(self._key_prefix + array_meta_key, b""))

        h.update(self.store.get(self.attrs.key, b""))

        checksum = h.digest()

        return checksum

    def hexdigest(self, hashname="sha1"):
        """
        Compute a checksum for the data. Default uses sha1 for speed.

        Examples
        --------
        >>> import zarr
        >>> z = zarr.empty(shape=(10000, 10000), chunks=(1000, 1000))
        >>> z.hexdigest()
        '041f90bc7a571452af4f850a8ca2c6cddfa8a1ac'
        >>> z = zarr.zeros(shape=(10000, 10000), chunks=(1000, 1000))
        >>> z.hexdigest()
        '7162d416d26a68063b66ed1f30e0a866e4abed60'
        >>> z = zarr.zeros(shape=(10000, 10000), dtype="u1", chunks=(1000, 1000))
        >>> z.hexdigest()
        'cb387af37410ae5a3222e893cf3373e4e4f22816'
        """

        checksum = binascii.hexlify(self.digest(hashname=hashname))

        # This is a bytes object on Python 3 and we want a str.
        if type(checksum) is not str:
            checksum = checksum.decode('utf8')

        return checksum

    def __getstate__(self):
        return (self._store, self._path, self._read_only, self._chunk_store,
                self._synchronizer, self._cache_metadata, self._attrs.cache,
                self._partial_decompress, self._write_empty_chunks)

    def __setstate__(self, state):
        self.__init__(*state)

    def _synchronized_op(self, f, *args, **kwargs):

        if self._synchronizer is None:
            # no synchronization
            lock = nolock

        else:
            # synchronize on the array
            mkey = self._key_prefix + array_meta_key
            lock = self._synchronizer[mkey]

        with lock:
            self._refresh_metadata_nosync()
            result = f(*args, **kwargs)

        return result

    def _write_op(self, f, *args, **kwargs):

        # guard condition
        if self._read_only:
            raise ReadOnlyError()

        return self._synchronized_op(f, *args, **kwargs)

    def resize(self, *args):
        """Change the shape of the array by growing or shrinking one or more
        dimensions.

        Examples
        --------
        >>> import zarr
        >>> z = zarr.zeros(shape=(10000, 10000), chunks=(1000, 1000))
        >>> z.shape
        (10000, 10000)
        >>> z.resize(20000, 10000)
        >>> z.shape
        (20000, 10000)
        >>> z.resize(30000, 1000)
        >>> z.shape
        (30000, 1000)

        Notes
        -----
        When resizing an array, the data are not rearranged in any way.

        If one or more dimensions are shrunk, any chunks falling outside the
        new array shape will be deleted from the underlying store.

        """

        return self._write_op(self._resize_nosync, *args)

    def _resize_nosync(self, *args):

        # normalize new shape argument
        old_shape = self._shape
        new_shape = normalize_resize_args(old_shape, *args)
        old_cdata_shape = self._cdata_shape

        # update metadata
        self._shape = new_shape
        self._flush_metadata_nosync()

        # determine the new number and arrangement of chunks
        chunks = self._chunks
        new_cdata_shape = tuple(math.ceil(s / c)
                                for s, c in zip(new_shape, chunks))

        # remove any chunks not within range
        chunk_store = self.chunk_store
        for cidx in itertools.product(*[range(n) for n in old_cdata_shape]):
            if all(i < c for i, c in zip(cidx, new_cdata_shape)):
                pass  # keep the chunk
            else:
                key = self._chunk_key(cidx)
                try:
                    del chunk_store[key]
                except KeyError:
                    # chunk not initialized
                    pass

    def append(self, data, axis=0):
        """Append `data` to `axis`.

        Parameters
        ----------
        data : array_like
            Data to be appended.
        axis : int
            Axis along which to append.

        Returns
        -------
        new_shape : tuple

        Notes
        -----
        The size of all dimensions other than `axis` must match between this
        array and `data`.

        Examples
        --------
        >>> import numpy as np
        >>> import zarr
        >>> a = np.arange(10000000, dtype='i4').reshape(10000, 1000)
        >>> z = zarr.array(a, chunks=(1000, 100))
        >>> z.shape
        (10000, 1000)
        >>> z.append(a)
        (20000, 1000)
        >>> z.append(np.vstack([a, a]), axis=1)
        (20000, 2000)
        >>> z.shape
        (20000, 2000)

        """
        return self._write_op(self._append_nosync, data, axis=axis)

    def _append_nosync(self, data, axis=0):

        # ensure data is array-like
        if not hasattr(data, 'shape'):
            data = np.asanyarray(data)

        # ensure shapes are compatible for non-append dimensions
        self_shape_preserved = tuple(s for i, s in enumerate(self._shape)
                                     if i != axis)
        data_shape_preserved = tuple(s for i, s in enumerate(data.shape)
                                     if i != axis)
        if self_shape_preserved != data_shape_preserved:
            raise ValueError('shape of data to append is not compatible with the array; '
                             'all dimensions must match except for the dimension being '
                             'appended')

        # remember old shape
        old_shape = self._shape

        # determine new shape
        new_shape = tuple(
            self._shape[i] if i != axis else self._shape[i] + data.shape[i]
            for i in range(len(self._shape))
        )

        # resize
        self._resize_nosync(new_shape)

        # store data
        # noinspection PyTypeChecker
        append_selection = tuple(
            slice(None) if i != axis else slice(old_shape[i], new_shape[i])
            for i in range(len(self._shape))
        )
        self[append_selection] = data

        return new_shape

    def view(self, shape=None, chunks=None, dtype=None,
             fill_value=None, filters=None, read_only=None,
             synchronizer=None):
        """Return an array sharing the same data.

        Parameters
        ----------
        shape : int or tuple of ints
            Array shape.
        chunks : int or tuple of ints, optional
            Chunk shape.
        dtype : string or dtype, optional
            NumPy dtype.
        fill_value : object
            Default value to use for uninitialized portions of the array.
        filters : sequence, optional
            Sequence of filters to use to encode chunk data prior to
            compression.
        read_only : bool, optional
            True if array should be protected against modification.
        synchronizer : object, optional
            Array synchronizer.

        Notes
        -----
        WARNING: This is an experimental feature and should be used with care.
        There are plenty of ways to generate errors and/or cause data
        corruption.

        Examples
        --------

        Bypass filters:

            >>> import zarr
            >>> import numpy as np
            >>> np.random.seed(42)
            >>> labels = ['female', 'male']
            >>> data = np.random.choice(labels, size=10000)
            >>> filters = [zarr.Categorize(labels=labels,
            ...                            dtype=data.dtype,
            ...                            astype='u1')]
            >>> a = zarr.array(data, chunks=1000, filters=filters)
            >>> a[:]
            array(['female', 'male', 'female', ..., 'male', 'male', 'female'],
                  dtype='<U6')
            >>> v = a.view(dtype='u1', filters=[])
            >>> v.is_view
            True
            >>> v[:]
            array([1, 2, 1, ..., 2, 2, 1], dtype=uint8)

        Views can be used to modify data:

            >>> x = v[:]
            >>> x.sort()
            >>> v[:] = x
            >>> v[:]
            array([1, 1, 1, ..., 2, 2, 2], dtype=uint8)
            >>> a[:]
            array(['female', 'female', 'female', ..., 'male', 'male', 'male'],
                  dtype='<U6')

        View as a different dtype with the same item size:

            >>> data = np.random.randint(0, 2, size=10000, dtype='u1')
            >>> a = zarr.array(data, chunks=1000)
            >>> a[:]
            array([0, 0, 1, ..., 1, 0, 0], dtype=uint8)
            >>> v = a.view(dtype=bool)
            >>> v[:]
            array([False, False,  True, ...,  True, False, False])
            >>> np.all(a[:].view(dtype=bool) == v[:])
            True

        An array can be viewed with a dtype with a different item size, however
        some care is needed to adjust the shape and chunk shape so that chunk
        data is interpreted correctly:

            >>> data = np.arange(10000, dtype='u2')
            >>> a = zarr.array(data, chunks=1000)
            >>> a[:10]
            array([0, 1, 2, 3, 4, 5, 6, 7, 8, 9], dtype=uint16)
            >>> v = a.view(dtype='u1', shape=20000, chunks=2000)
            >>> v[:10]
            array([0, 0, 1, 0, 2, 0, 3, 0, 4, 0], dtype=uint8)
            >>> np.all(a[:].view('u1') == v[:])
            True

        Change fill value for uninitialized chunks:

            >>> a = zarr.full(10000, chunks=1000, fill_value=-1, dtype='i1')
            >>> a[:]
            array([-1, -1, -1, ..., -1, -1, -1], dtype=int8)
            >>> v = a.view(fill_value=42)
            >>> v[:]
            array([42, 42, 42, ..., 42, 42, 42], dtype=int8)

        Note that resizing or appending to views is not permitted:

            >>> a = zarr.empty(10000)
            >>> v = a.view()
            >>> try:
            ...     v.resize(20000)
            ... except PermissionError as e:
            ...     print(e)
            operation not permitted for views

        """

        store = self._store
        chunk_store = self._chunk_store
        path = self._path
        if read_only is None:
            read_only = self._read_only
        if synchronizer is None:
            synchronizer = self._synchronizer
        a = Array(store=store, path=path, chunk_store=chunk_store, read_only=read_only,
                  synchronizer=synchronizer, cache_metadata=True)
        a._is_view = True

        # allow override of some properties
        if dtype is None:
            dtype = self._dtype
        else:
            dtype = np.dtype(dtype)
            a._dtype = dtype
        if shape is None:
            shape = self._shape
        else:
            shape = normalize_shape(shape)
            a._shape = shape
        if chunks is not None:
            chunks = normalize_chunks(chunks, shape, dtype.itemsize)
            a._chunks = chunks
        if fill_value is not None:
            a._fill_value = fill_value
        if filters is not None:
            a._filters = filters

        return a

    def astype(self, dtype):
        """Returns a view that does on the fly type conversion of the underlying data.

        Parameters
        ----------
        dtype : string or dtype
            NumPy dtype.

        Notes
        -----
        This method returns a new Array object which is a view on the same
        underlying chunk data. Modifying any data via the view is currently
        not permitted and will result in an error. This is an experimental
        feature and its behavior is subject to change in the future.

        See Also
        --------
        Array.view

        Examples
        --------

        >>> import zarr
        >>> import numpy as np
        >>> data = np.arange(100, dtype=np.uint8)
        >>> a = zarr.array(data, chunks=10)
        >>> a[:]
        array([ 0,  1,  2,  3,  4,  5,  6,  7,  8,  9, 10, 11, 12, 13, 14, 15,
               16, 17, 18, 19, 20, 21, 22, 23, 24, 25, 26, 27, 28, 29, 30, 31,
               32, 33, 34, 35, 36, 37, 38, 39, 40, 41, 42, 43, 44, 45, 46, 47,
               48, 49, 50, 51, 52, 53, 54, 55, 56, 57, 58, 59, 60, 61, 62, 63,
               64, 65, 66, 67, 68, 69, 70, 71, 72, 73, 74, 75, 76, 77, 78, 79,
               80, 81, 82, 83, 84, 85, 86, 87, 88, 89, 90, 91, 92, 93, 94, 95,
               96, 97, 98, 99], dtype=uint8)
        >>> v = a.astype(np.float32)
        >>> v.is_view
        True
        >>> v[:]
        array([  0.,   1.,   2.,   3.,   4.,   5.,   6.,   7.,   8.,   9.,
                10.,  11.,  12.,  13.,  14.,  15.,  16.,  17.,  18.,  19.,
                20.,  21.,  22.,  23.,  24.,  25.,  26.,  27.,  28.,  29.,
                30.,  31.,  32.,  33.,  34.,  35.,  36.,  37.,  38.,  39.,
                40.,  41.,  42.,  43.,  44.,  45.,  46.,  47.,  48.,  49.,
                50.,  51.,  52.,  53.,  54.,  55.,  56.,  57.,  58.,  59.,
                60.,  61.,  62.,  63.,  64.,  65.,  66.,  67.,  68.,  69.,
                70.,  71.,  72.,  73.,  74.,  75.,  76.,  77.,  78.,  79.,
                80.,  81.,  82.,  83.,  84.,  85.,  86.,  87.,  88.,  89.,
                90.,  91.,  92.,  93.,  94.,  95.,  96.,  97.,  98.,  99.],
              dtype=float32)
        """

        dtype = np.dtype(dtype)

        filters = []
        if self._filters:
            filters.extend(self._filters)
        filters.insert(0, AsType(encode_dtype=self._dtype, decode_dtype=dtype))

        return self.view(filters=filters, dtype=dtype, read_only=True) class CopyError(RuntimeError):
    pass class Group(MutableMapping):
    """Instantiate a group from an initialized store.

    Parameters
    ----------
    store : MutableMapping
        Group store, already initialized.
        If the Group is used in a context manager, and the store has a ``close`` method,
        it will be called on exit.
    path : string, optional
        Group path.
    read_only : bool, optional
        True if group should be protected against modification.
    chunk_store : MutableMapping, optional
        Separate storage for chunks. If not provided, `store` will be used
        for storage of both chunks and metadata.
    cache_attrs : bool, optional
        If True (default), user attributes will be cached for attribute read
        operations. If False, user attributes are reloaded from the store prior
        to all attribute read operations.
    synchronizer : object, optional
        Array synchronizer.

    Attributes
    ----------
    store
    path
    name
    read_only
    chunk_store
    synchronizer
    attrs
    info

    Methods
    -------
    __len__
    __iter__
    __contains__
    __getitem__
    __enter__
    __exit__
    group_keys
    groups
    array_keys
    arrays
    visit
    visitkeys
    visitvalues
    visititems
    tree
    create_group
    require_group
    create_groups
    require_groups
    create_dataset
    require_dataset
    create
    empty
    zeros
    ones
    full
    array
    empty_like
    zeros_like
    ones_like
    full_like
    info
    move

    """

    def __init__(self, store, path=None, read_only=False, chunk_store=None,
                 cache_attrs=True, synchronizer=None):
        store: BaseStore = BaseStore._ensure_store(store)
        chunk_store: BaseStore = BaseStore._ensure_store(chunk_store)
        self._store = store
        self._chunk_store = chunk_store
        self._path = normalize_storage_path(path)
        if self._path:
            self._key_prefix = self._path + '/'
        else:
            self._key_prefix = ''
        self._read_only = read_only
        self._synchronizer = synchronizer

        # guard conditions
        if contains_array(store, path=self._path):
            raise ContainsArrayError(path)

        # initialize metadata
        try:
            mkey = self._key_prefix + group_meta_key
            meta_bytes = store[mkey]
        except KeyError:
            raise GroupNotFoundError(path)
        else:
            self._meta = self._store._metadata_class.decode_group_metadata(meta_bytes)

        # setup attributes
        akey = self._key_prefix + attrs_key
        self._attrs = Attributes(store, key=akey, read_only=read_only,
                                 cache=cache_attrs, synchronizer=synchronizer)

        # setup info
        self._info = InfoReporter(self)

    @property
    def store(self):
        """A MutableMapping providing the underlying storage for the group."""
        return self._store

    @property
    def path(self):
        """Storage path."""
        return self._path

    @property
    def name(self):
        """Group name following h5py convention."""
        if self._path:
            # follow h5py convention: add leading slash
            name = self._path
            if name[0] != '/':
                name = '/' + name
            return name
        return '/'

    @property
    def basename(self):
        """Final component of name."""
        return self.name.split('/')[-1]

    @property
    def read_only(self):
        """A boolean, True if modification operations are not permitted."""
        return self._read_only

    @property
    def chunk_store(self):
        """A MutableMapping providing the underlying storage for array chunks."""
        if self._chunk_store is None:
            return self._store
        else:
            return self._chunk_store

    @property
    def synchronizer(self):
        """Object used to synchronize write access to groups and arrays."""
        return self._synchronizer

    @property
    def attrs(self):
        """A MutableMapping containing user-defined attributes. Note that
        attribute values must be JSON serializable."""
        return self._attrs

    @property
    def info(self):
        """Return diagnostic information about the group."""
        return self._info

    def __eq__(self, other):
        return (
            isinstance(other, Group) and
            self._store == other.store and
            self._read_only == other.read_only and
            self._path == other.path
            # N.B., no need to compare attributes, should be covered by
            # store comparison
        )

    def __iter__(self):
        """Return an iterator over group member names.

        Examples
        --------
        >>> import zarr
        >>> g1 = zarr.group()
        >>> g2 = g1.create_group('foo')
        >>> g3 = g1.create_group('bar')
        >>> d1 = g1.create_dataset('baz', shape=100, chunks=10)
        >>> d2 = g1.create_dataset('quux', shape=200, chunks=20)
        >>> for name in g1:
        ...     print(name)
        bar
        baz
        foo
        quux

        """
        for key in sorted(listdir(self._store, self._path)):
            path = self._key_prefix + key
            if (contains_array(self._store, path) or
                    contains_group(self._store, path)):
                yield key

    def __len__(self):
        """Number of members."""
        return sum(1 for _ in self)

    def __repr__(self):
        t = type(self)
        r = '<{}.{}'.format(t.__module__, t.__name__)
        if self.name:
            r += ' %r' % self.name
        if self._read_only:
            r += ' read-only'
        r += '>'
        return r

    def __enter__(self):
        """Return the Group for use as a context manager."""
        return self

    def __exit__(self, exc_type, exc_val, exc_tb):
        """Call the close method of the underlying Store."""
        self.store.close()

    def info_items(self):

        def typestr(o):
            return '{}.{}'.format(type(o).__module__, type(o).__name__)

        items = []

        # basic info
        if self.name is not None:
            items += [('Name', self.name)]
        items += [
            ('Type', typestr(self)),
            ('Read-only', str(self.read_only)),
        ]

        # synchronizer
        if self._synchronizer is not None:
            items += [('Synchronizer type', typestr(self._synchronizer))]

        # storage info
        items += [('Store type', typestr(self._store))]
        if self._chunk_store is not None:
            items += [('Chunk store type', typestr(self._chunk_store))]

        # members
        items += [('No. members', len(self))]
        array_keys = sorted(self.array_keys())
        group_keys = sorted(self.group_keys())
        items += [('No. arrays', len(array_keys))]
        items += [('No. groups', len(group_keys))]
        if array_keys:
            items += [('Arrays', ', '.join(array_keys))]
        if group_keys:
            items += [('Groups', ', '.join(group_keys))]

        return items

    def __getstate__(self):
        return (self._store, self._path, self._read_only, self._chunk_store,
                self.attrs.cache, self._synchronizer)

    def __setstate__(self, state):
        self.__init__(*state)

    def _item_path(self, item):
        absolute = isinstance(item, str) and item and item[0] == '/'
        path = normalize_storage_path(item)
        if not absolute and self._path:
            path = self._key_prefix + path
        return path

    def __contains__(self, item):
        """Test for group membership.

        Examples
        --------
        >>> import zarr
        >>> g1 = zarr.group()
        >>> g2 = g1.create_group('foo')
        >>> d1 = g1.create_dataset('bar', shape=100, chunks=10)
        >>> 'foo' in g1
        True
        >>> 'bar' in g1
        True
        >>> 'baz' in g1
        False

        """
        path = self._item_path(item)
        return contains_array(self._store, path) or \
            contains_group(self._store, path)

    def __getitem__(self, item):
        """Obtain a group member.

        Parameters
        ----------
        item : string
            Member name or path.

        Examples
        --------
        >>> import zarr
        >>> g1 = zarr.group()
        >>> d1 = g1.create_dataset('foo/bar/baz', shape=100, chunks=10)
        >>> g1['foo']
        <zarr.hierarchy.Group '/foo'>
        >>> g1['foo/bar']
        <zarr.hierarchy.Group '/foo/bar'>
        >>> g1['foo/bar/baz']
        <zarr.core.Array '/foo/bar/baz' (100,) float64>

        """
        path = self._item_path(item)
        if contains_array(self._store, path):
            return Array(self._store, read_only=self._read_only, path=path,
                         chunk_store=self._chunk_store,
                         synchronizer=self._synchronizer, cache_attrs=self.attrs.cache)
        elif contains_group(self._store, path):
            return Group(self._store, read_only=self._read_only, path=path,
                         chunk_store=self._chunk_store, cache_attrs=self.attrs.cache,
                         synchronizer=self._synchronizer)
        else:
            raise KeyError(item)

    def __setitem__(self, item, value):
        self.array(item, value, overwrite=True)

    def __delitem__(self, item):
        return self._write_op(self._delitem_nosync, item)

    def _delitem_nosync(self, item):
        path = self._item_path(item)
        if contains_array(self._store, path) or \
                contains_group(self._store, path):
            rmdir(self._store, path)
        else:
            raise KeyError(item)

    def __getattr__(self, item):
        # allow access to group members via dot notation
        try:
            return self.__getitem__(item)
        except KeyError:
            raise AttributeError

    def __dir__(self):
        # noinspection PyUnresolvedReferences
        base = super().__dir__()
        keys = sorted(set(base + list(self)))
        keys = [k for k in keys if is_valid_python_name(k)]
        return keys

    def _ipython_key_completions_(self):
        return sorted(self)

    def group_keys(self):
        """Return an iterator over member names for groups only.

        Examples
        --------
        >>> import zarr
        >>> g1 = zarr.group()
        >>> g2 = g1.create_group('foo')
        >>> g3 = g1.create_group('bar')
        >>> d1 = g1.create_dataset('baz', shape=100, chunks=10)
        >>> d2 = g1.create_dataset('quux', shape=200, chunks=20)
        >>> sorted(g1.group_keys())
        ['bar', 'foo']

        """
        for key in sorted(listdir(self._store, self._path)):
            path = self._key_prefix + key
            if contains_group(self._store, path):
                yield key

    def groups(self):
        """Return an iterator over (name, value) pairs for groups only.

        Examples
        --------
        >>> import zarr
        >>> g1 = zarr.group()
        >>> g2 = g1.create_group('foo')
        >>> g3 = g1.create_group('bar')
        >>> d1 = g1.create_dataset('baz', shape=100, chunks=10)
        >>> d2 = g1.create_dataset('quux', shape=200, chunks=20)
        >>> for n, v in g1.groups():
        ...     print(n, type(v))
        bar <class 'zarr.hierarchy.Group'>
        foo <class 'zarr.hierarchy.Group'>

        """
        for key in sorted(listdir(self._store, self._path)):
            path = self._key_prefix + key
            if contains_group(self._store, path):
                yield key, Group(self._store, path=path, read_only=self._read_only,
                                 chunk_store=self._chunk_store,
                                 cache_attrs=self.attrs.cache,
                                 synchronizer=self._synchronizer)

    def array_keys(self, recurse=False):
        """Return an iterator over member names for arrays only.

        Parameters
        ----------
        recurse : recurse, optional
            Option to return member names for all arrays, even from groups
            below the current one. If False, only member names for arrays in
            the current group will be returned. Default value is False.

        Examples
        --------
        >>> import zarr
        >>> g1 = zarr.group()
        >>> g2 = g1.create_group('foo')
        >>> g3 = g1.create_group('bar')
        >>> d1 = g1.create_dataset('baz', shape=100, chunks=10)
        >>> d2 = g1.create_dataset('quux', shape=200, chunks=20)
        >>> sorted(g1.array_keys())
        ['baz', 'quux']

        """
        return self._array_iter(keys_only=True,
                                method='array_keys',
                                recurse=recurse)

    def arrays(self, recurse=False):
        """Return an iterator over (name, value) pairs for arrays only.

        Parameters
        ----------
        recurse : recurse, optional
            Option to return (name, value) pairs for all arrays, even from groups
            below the current one. If False, only (name, value) pairs for arrays in
            the current group will be returned. Default value is False.

        Examples
        --------
        >>> import zarr
        >>> g1 = zarr.group()
        >>> g2 = g1.create_group('foo')
        >>> g3 = g1.create_group('bar')
        >>> d1 = g1.create_dataset('baz', shape=100, chunks=10)
        >>> d2 = g1.create_dataset('quux', shape=200, chunks=20)
        >>> for n, v in g1.arrays():
        ...     print(n, type(v))
        baz <class 'zarr.core.Array'>
        quux <class 'zarr.core.Array'>

        """
        return self._array_iter(keys_only=False,
                                method='arrays',
                                recurse=recurse)

    def _array_iter(self, keys_only, method, recurse):
        for key in sorted(listdir(self._store, self._path)):
            path = self._key_prefix + key
            if contains_array(self._store, path):
                yield key if keys_only else (key, self[key])
            elif recurse and contains_group(self._store, path):
                group = self[key]
                for i in getattr(group, method)(recurse=recurse):
                    yield i

    def visitvalues(self, func):
        """Run ``func`` on each object.

        Note: If ``func`` returns ``None`` (or doesn't return),
              iteration continues. However, if ``func`` returns
              anything else, it ceases and returns that value.

        Examples
        --------
        >>> import zarr
        >>> g1 = zarr.group()
        >>> g2 = g1.create_group('foo')
        >>> g3 = g1.create_group('bar')
        >>> g4 = g3.create_group('baz')
        >>> g5 = g3.create_group('quux')
        >>> def print_visitor(obj):
        ...     print(obj)
        >>> g1.visitvalues(print_visitor)
        <zarr.hierarchy.Group '/bar'>
        <zarr.hierarchy.Group '/bar/baz'>
        <zarr.hierarchy.Group '/bar/quux'>
        <zarr.hierarchy.Group '/foo'>
        >>> g3.visitvalues(print_visitor)
        <zarr.hierarchy.Group '/bar/baz'>
        <zarr.hierarchy.Group '/bar/quux'>

        """

        def _visit(obj):
            yield obj
            keys = sorted(getattr(obj, "keys", lambda: [])())
            for k in keys:
                for v in _visit(obj[k]):
                    yield v

        for each_obj in islice(_visit(self), 1, None):
            value = func(each_obj)
            if value is not None:
                return value

    def visit(self, func):
        """Run ``func`` on each object's path.

        Note: If ``func`` returns ``None`` (or doesn't return),
              iteration continues. However, if ``func`` returns
              anything else, it ceases and returns that value.

        Examples
        --------
        >>> import zarr
        >>> g1 = zarr.group()
        >>> g2 = g1.create_group('foo')
        >>> g3 = g1.create_group('bar')
        >>> g4 = g3.create_group('baz')
        >>> g5 = g3.create_group('quux')
        >>> def print_visitor(name):
        ...     print(name)
        >>> g1.visit(print_visitor)
        bar
        bar/baz
        bar/quux
        foo
        >>> g3.visit(print_visitor)
        baz
        quux

        """

        base_len = len(self.name)
        return self.visitvalues(lambda o: func(o.name[base_len:].lstrip("/")))

    def visitkeys(self, func):
        """An alias for :py:meth:`~Group.visit`.
        """

        return self.visit(func)

    def visititems(self, func):
        """Run ``func`` on each object's path and the object itself.

        Note: If ``func`` returns ``None`` (or doesn't return),
              iteration continues. However, if ``func`` returns
              anything else, it ceases and returns that value.

        Examples
        --------
        >>> import zarr
        >>> g1 = zarr.group()
        >>> g2 = g1.create_group('foo')
        >>> g3 = g1.create_group('bar')
        >>> g4 = g3.create_group('baz')
        >>> g5 = g3.create_group('quux')
        >>> def print_visitor(name, obj):
        ...     print((name, obj))
        >>> g1.visititems(print_visitor)
        ('bar', <zarr.hierarchy.Group '/bar'>)
        ('bar/baz', <zarr.hierarchy.Group '/bar/baz'>)
        ('bar/quux', <zarr.hierarchy.Group '/bar/quux'>)
        ('foo', <zarr.hierarchy.Group '/foo'>)
        >>> g3.visititems(print_visitor)
        ('baz', <zarr.hierarchy.Group '/bar/baz'>)
        ('quux', <zarr.hierarchy.Group '/bar/quux'>)

        """

        base_len = len(self.name)
        return self.visitvalues(lambda o: func(o.name[base_len:].lstrip("/"), o))

    def tree(self, expand=False, level=None):
        """Provide a ``print``-able display of the hierarchy.

        Parameters
        ----------
        expand : bool, optional
            Only relevant for HTML representation. If True, tree will be fully expanded.
        level : int, optional
            Maximum depth to descend into hierarchy.

        Examples
        --------
        >>> import zarr
        >>> g1 = zarr.group()
        >>> g2 = g1.create_group('foo')
        >>> g3 = g1.create_group('bar')
        >>> g4 = g3.create_group('baz')
        >>> g5 = g3.create_group('quux')
        >>> d1 = g5.create_dataset('baz', shape=100, chunks=10)
        >>> g1.tree()
        /
         ├── bar
         │   ├── baz
         │   └── quux
         │       └── baz (100,) float64
         └── foo
        >>> g1.tree(level=2)
        /
         ├── bar
         │   ├── baz
         │   └── quux
         └── foo
        >>> g3.tree()
        bar
         ├── baz
         └── quux
             └── baz (100,) float64

        Notes
        -----
        Please note that this is an experimental feature. The behaviour of this
        function is still evolving and the default output and/or parameters may change
        in future versions.

        """

        return TreeViewer(self, expand=expand, level=level)

    def _write_op(self, f, *args, **kwargs):

        # guard condition
        if self._read_only:
            raise ReadOnlyError()

        if self._synchronizer is None:
            # no synchronization
            lock = nolock
        else:
            # synchronize on the root group
            lock = self._synchronizer[group_meta_key]

        with lock:
            return f(*args, **kwargs)

    def create_group(self, name, overwrite=False):
        """Create a sub-group.

        Parameters
        ----------
        name : string
            Group name.
        overwrite : bool, optional
            If True, overwrite any existing array with the given name.

        Returns
        -------
        g : zarr.hierarchy.Group

        Examples
        --------
        >>> import zarr
        >>> g1 = zarr.group()
        >>> g2 = g1.create_group('foo')
        >>> g3 = g1.create_group('bar')
        >>> g4 = g1.create_group('baz/quux')

        """

        return self._write_op(self._create_group_nosync, name, overwrite=overwrite)

    def _create_group_nosync(self, name, overwrite=False):
        path = self._item_path(name)

        # create terminal group
        init_group(self._store, path=path, chunk_store=self._chunk_store,
                   overwrite=overwrite)

        return Group(self._store, path=path, read_only=self._read_only,
                     chunk_store=self._chunk_store, cache_attrs=self.attrs.cache,
                     synchronizer=self._synchronizer)

    def create_groups(self, *names, **kwargs):
        """Convenience method to create multiple groups in a single call."""
        return tuple(self.create_group(name, **kwargs) for name in names)

    def require_group(self, name, overwrite=False):
        """Obtain a sub-group, creating one if it doesn't exist.

        Parameters
        ----------
        name : string
            Group name.
        overwrite : bool, optional
            Overwrite any existing array with given `name` if present.

        Returns
        -------
        g : zarr.hierarchy.Group

        Examples
        --------
        >>> import zarr
        >>> g1 = zarr.group()
        >>> g2 = g1.require_group('foo')
        >>> g3 = g1.require_group('foo')
        >>> g2 == g3
        True

        """

        return self._write_op(self._require_group_nosync, name,
                              overwrite=overwrite)

    def _require_group_nosync(self, name, overwrite=False):
        path = self._item_path(name)

        # create terminal group if necessary
        if not contains_group(self._store, path):
            init_group(store=self._store, path=path, chunk_store=self._chunk_store,
                       overwrite=overwrite)

        return Group(self._store, path=path, read_only=self._read_only,
                     chunk_store=self._chunk_store, cache_attrs=self.attrs.cache,
                     synchronizer=self._synchronizer)

    def require_groups(self, *names):
        """Convenience method to require multiple groups in a single call."""
        return tuple(self.require_group(name) for name in names)

    # noinspection PyIncorrectDocstring
    def create_dataset(self, name, **kwargs):
        """Create an array.

        Arrays are known as "datasets" in HDF5 terminology. For compatibility
        with h5py, Zarr groups also implement the require_dataset() method.

        Parameters
        ----------
        name : string
            Array name.
        data : array_like, optional
            Initial data.
        shape : int or tuple of ints
            Array shape.
        chunks : int or tuple of ints, optional
            Chunk shape. If not provided, will be guessed from `shape` and
            `dtype`.
        dtype : string or dtype, optional
            NumPy dtype.
        compressor : Codec, optional
            Primary compressor.
        fill_value : object
            Default value to use for uninitialized portions of the array.
        order : {'C', 'F'}, optional
            Memory layout to be used within each chunk.
        synchronizer : zarr.sync.ArraySynchronizer, optional
            Array synchronizer.
        filters : sequence of Codecs, optional
            Sequence of filters to use to encode chunk data prior to
            compression.
        overwrite : bool, optional
            If True, replace any existing array or group with the given name.
        cache_metadata : bool, optional
            If True, array configuration metadata will be cached for the
            lifetime of the object. If False, array metadata will be reloaded
            prior to all data access and modification operations (may incur
            overhead depending on storage and data access pattern).
        dimension_separator : {'.', '/'}, optional
            Separator placed between the dimensions of a chunk.

        Returns
        -------
        a : zarr.core.Array

        Examples
        --------
        >>> import zarr
        >>> g1 = zarr.group()
        >>> d1 = g1.create_dataset('foo', shape=(10000, 10000),
        ...                        chunks=(1000, 1000))
        >>> d1
        <zarr.core.Array '/foo' (10000, 10000) float64>
        >>> d2 = g1.create_dataset('bar/baz/qux', shape=(100, 100, 100),
        ...                        chunks=(100, 10, 10))
        >>> d2
        <zarr.core.Array '/bar/baz/qux' (100, 100, 100) float64>

        """
        assert "mode" not in kwargs

        return self._write_op(self._create_dataset_nosync, name, **kwargs)

    def _create_dataset_nosync(self, name, data=None, **kwargs):

        assert "mode" not in kwargs
        path = self._item_path(name)

        # determine synchronizer
        kwargs.setdefault('synchronizer', self._synchronizer)
        kwargs.setdefault('cache_attrs', self.attrs.cache)

        # create array
        if data is None:
            a = create(store=self._store, path=path, chunk_store=self._chunk_store,
                       **kwargs)

        else:
            a = array(data, store=self._store, path=path, chunk_store=self._chunk_store,
                      **kwargs)

        return a

    def require_dataset(self, name, shape, dtype=None, exact=False, **kwargs):
        """Obtain an array, creating if it doesn't exist.

        Arrays are known as "datasets" in HDF5 terminology. For compatibility
        with h5py, Zarr groups also implement the create_dataset() method.

        Other `kwargs` are as per :func:`zarr.hierarchy.Group.create_dataset`.

        Parameters
        ----------
        name : string
            Array name.
        shape : int or tuple of ints
            Array shape.
        dtype : string or dtype, optional
            NumPy dtype.
        exact : bool, optional
            If True, require `dtype` to match exactly. If false, require
            `dtype` can be cast from array dtype.

        """

        return self._write_op(self._require_dataset_nosync, name, shape=shape,
                              dtype=dtype, exact=exact, **kwargs)

    def _require_dataset_nosync(self, name, shape, dtype=None, exact=False,
                                **kwargs):

        path = self._item_path(name)

        if contains_array(self._store, path):

            # array already exists at path, validate that it is the right shape and type

            synchronizer = kwargs.get('synchronizer', self._synchronizer)
            cache_metadata = kwargs.get('cache_metadata', True)
            cache_attrs = kwargs.get('cache_attrs', self.attrs.cache)
            a = Array(self._store, path=path, read_only=self._read_only,
                      chunk_store=self._chunk_store, synchronizer=synchronizer,
                      cache_metadata=cache_metadata, cache_attrs=cache_attrs)
            shape = normalize_shape(shape)
            if shape != a.shape:
                raise TypeError('shape do not match existing array; expected {}, got {}'
                                .format(a.shape, shape))
            dtype = np.dtype(dtype)
            if exact:
                if dtype != a.dtype:
                    raise TypeError('dtypes do not match exactly; expected {}, got {}'
                                    .format(a.dtype, dtype))
            else:
                if not np.can_cast(dtype, a.dtype):
                    raise TypeError('dtypes ({}, {}) cannot be safely cast'
                                    .format(dtype, a.dtype))
            return a

        else:
            return self._create_dataset_nosync(name, shape=shape, dtype=dtype,
                                               **kwargs)

    def create(self, name, **kwargs):
        """Create an array. Keyword arguments as per
        :func:`zarr.creation.create`."""
        return self._write_op(self._create_nosync, name, **kwargs)

    def _create_nosync(self, name, **kwargs):
        path = self._item_path(name)
        kwargs.setdefault('synchronizer', self._synchronizer)
        kwargs.setdefault('cache_attrs', self.attrs.cache)
        return create(store=self._store, path=path, chunk_store=self._chunk_store,
                      **kwargs)

    def empty(self, name, **kwargs):
        """Create an array. Keyword arguments as per
        :func:`zarr.creation.empty`."""
        return self._write_op(self._empty_nosync, name, **kwargs)

    def _empty_nosync(self, name, **kwargs):
        path = self._item_path(name)
        kwargs.setdefault('synchronizer', self._synchronizer)
        kwargs.setdefault('cache_attrs', self.attrs.cache)
        return empty(store=self._store, path=path, chunk_store=self._chunk_store,
                     **kwargs)

    def zeros(self, name, **kwargs):
        """Create an array. Keyword arguments as per
        :func:`zarr.creation.zeros`."""
        return self._write_op(self._zeros_nosync, name, **kwargs)

    def _zeros_nosync(self, name, **kwargs):
        path = self._item_path(name)
        kwargs.setdefault('synchronizer', self._synchronizer)
        kwargs.setdefault('cache_attrs', self.attrs.cache)
        return zeros(store=self._store, path=path, chunk_store=self._chunk_store,
                     **kwargs)

    def ones(self, name, **kwargs):
        """Create an array. Keyword arguments as per
        :func:`zarr.creation.ones`."""
        return self._write_op(self._ones_nosync, name, **kwargs)

    def _ones_nosync(self, name, **kwargs):
        path = self._item_path(name)
        kwargs.setdefault('synchronizer', self._synchronizer)
        kwargs.setdefault('cache_attrs', self.attrs.cache)
        return ones(store=self._store, path=path, chunk_store=self._chunk_store, **kwargs)

    def full(self, name, fill_value, **kwargs):
        """Create an array. Keyword arguments as per
        :func:`zarr.creation.full`."""
        return self._write_op(self._full_nosync, name, fill_value, **kwargs)

    def _full_nosync(self, name, fill_value, **kwargs):
        path = self._item_path(name)
        kwargs.setdefault('synchronizer', self._synchronizer)
        kwargs.setdefault('cache_attrs', self.attrs.cache)
        return full(store=self._store, path=path, chunk_store=self._chunk_store,
                    fill_value=fill_value, **kwargs)

    def array(self, name, data, **kwargs):
        """Create an array. Keyword arguments as per
        :func:`zarr.creation.array`."""
        return self._write_op(self._array_nosync, name, data, **kwargs)

    def _array_nosync(self, name, data, **kwargs):
        path = self._item_path(name)
        kwargs.setdefault('synchronizer', self._synchronizer)
        kwargs.setdefault('cache_attrs', self.attrs.cache)
        return array(data, store=self._store, path=path, chunk_store=self._chunk_store,
                     **kwargs)

    def empty_like(self, name, data, **kwargs):
        """Create an array. Keyword arguments as per
        :func:`zarr.creation.empty_like`."""
        return self._write_op(self._empty_like_nosync, name, data, **kwargs)

    def _empty_like_nosync(self, name, data, **kwargs):
        path = self._item_path(name)
        kwargs.setdefault('synchronizer', self._synchronizer)
        kwargs.setdefault('cache_attrs', self.attrs.cache)
        return empty_like(data, store=self._store, path=path,
                          chunk_store=self._chunk_store, **kwargs)

    def zeros_like(self, name, data, **kwargs):
        """Create an array. Keyword arguments as per
        :func:`zarr.creation.zeros_like`."""
        return self._write_op(self._zeros_like_nosync, name, data, **kwargs)

    def _zeros_like_nosync(self, name, data, **kwargs):
        path = self._item_path(name)
        kwargs.setdefault('synchronizer', self._synchronizer)
        kwargs.setdefault('cache_attrs', self.attrs.cache)
        return zeros_like(data, store=self._store, path=path,
                          chunk_store=self._chunk_store, **kwargs)

    def ones_like(self, name, data, **kwargs):
        """Create an array. Keyword arguments as per
        :func:`zarr.creation.ones_like`."""
        return self._write_op(self._ones_like_nosync, name, data, **kwargs)

    def _ones_like_nosync(self, name, data, **kwargs):
        path = self._item_path(name)
        kwargs.setdefault('synchronizer', self._synchronizer)
        kwargs.setdefault('cache_attrs', self.attrs.cache)
        return ones_like(data, store=self._store, path=path,
                         chunk_store=self._chunk_store, **kwargs)

    def full_like(self, name, data, **kwargs):
        """Create an array. Keyword arguments as per
        :func:`zarr.creation.full_like`."""
        return self._write_op(self._full_like_nosync, name, data, **kwargs)

    def _full_like_nosync(self, name, data, **kwargs):
        path = self._item_path(name)
        kwargs.setdefault('synchronizer', self._synchronizer)
        kwargs.setdefault('cache_attrs', self.attrs.cache)
        return full_like(data, store=self._store, path=path,
                         chunk_store=self._chunk_store, **kwargs)

    def _move_nosync(self, path, new_path):
        rename(self._store, path, new_path)
        if self._chunk_store is not None:
            rename(self._chunk_store, path, new_path)

    def move(self, source, dest):
        """Move contents from one path to another relative to the Group.

        Parameters
        ----------
        source : string
            Name or path to a Zarr object to move.
        dest : string
            New name or path of the Zarr object.
        """

        source = self._item_path(source)
        dest = self._item_path(dest)

        # Check that source exists.
        if not (contains_array(self._store, source) or
                contains_group(self._store, source)):
            raise ValueError('The source, "%s", does not exist.' % source)
        if contains_array(self._store, dest) or contains_group(self._store, dest):
            raise ValueError('The dest, "%s", already exists.' % dest)

        # Ensure groups needed for `dest` exist.
        if "/" in dest:
            self.require_group("/" + dest.rsplit("/", 1)[0])

        self._write_op(self._move_nosync, source, dest) def group(store=None, overwrite=False, chunk_store=None,
          cache_attrs=True, synchronizer=None, path=None):
    """Create a group.

    Parameters
    ----------
    store : MutableMapping or string, optional
        Store or path to directory in file system.
    overwrite : bool, optional
        If True, delete any pre-existing data in `store` at `path` before
        creating the group.
    chunk_store : MutableMapping, optional
        Separate storage for chunks. If not provided, `store` will be used
        for storage of both chunks and metadata.
    cache_attrs : bool, optional
        If True (default), user attributes will be cached for attribute read
        operations. If False, user attributes are reloaded from the store prior
        to all attribute read operations.
    synchronizer : object, optional
        Array synchronizer.
    path : string, optional
        Group path within store.

    Returns
    -------
    g : zarr.hierarchy.Group

    Examples
    --------
    Create a group in memory::

        >>> import zarr
        >>> g = zarr.group()
        >>> g
        <zarr.hierarchy.Group '/'>

    Create a group with a different store::

        >>> store = zarr.DirectoryStore('data/example.zarr')
        >>> g = zarr.group(store=store, overwrite=True)
        >>> g
        <zarr.hierarchy.Group '/'>

    """

    # handle polymorphic store arg
    store = _normalize_store_arg(store)
    path = normalize_storage_path(path)

    # require group
    if overwrite or not contains_group(store):
        init_group(store, overwrite=overwrite, chunk_store=chunk_store,
                   path=path)

    return Group(store, read_only=False, chunk_store=chunk_store,
                 cache_attrs=cache_attrs, synchronizer=synchronizer, path=path) class ConsolidatedMetadataStore(Store):
    """A layer over other storage, where the metadata has been consolidated into
    a single key.

    The purpose of this class, is to be able to get all of the metadata for
    a given array in a single read operation from the underlying storage.
    See :func:`zarr.convenience.consolidate_metadata` for how to create this
    single metadata key.

    This class loads from the one key, and stores the data in a dict, so that
    accessing the keys no longer requires operations on the backend store.

    This class is read-only, and attempts to change the array metadata will
    fail, but changing the data is possible. If the backend storage is changed
    directly, then the metadata stored here could become obsolete, and
    :func:`zarr.convenience.consolidate_metadata` should be called again and the class
    re-invoked. The use case is for write once, read many times.

    .. versionadded:: 2.3

    .. note:: This is an experimental feature.

    Parameters
    ----------
    store: Store
        Containing the zarr array.
    metadata_key: str
        The target in the store where all of the metadata are stored. We
        assume JSON encoding.

    See Also
    --------
    zarr.convenience.consolidate_metadata, zarr.convenience.open_consolidated

    """

    def __init__(self, store: StoreLike, metadata_key=".zmetadata"):
        self.store = Store._ensure_store(store)

        # retrieve consolidated metadata
        meta = json_loads(store[metadata_key])

        # check format of consolidated metadata
        consolidated_format = meta.get('zarr_consolidated_format', None)
        if consolidated_format != 1:
            raise MetadataError('unsupported zarr consolidated metadata format: %s' %
                                consolidated_format)

        # decode metadata
        self.meta_store: Store = KVStore(meta["metadata"])

    def __getitem__(self, key):
        return self.meta_store[key]

    def __contains__(self, item):
        return item in self.meta_store

    def __iter__(self):
        return iter(self.meta_store)

    def __len__(self):
        return len(self.meta_store)

    def __delitem__(self, key):
        raise ReadOnlyError()

    def __setitem__(self, key, value):
        raise ReadOnlyError()

    def getsize(self, path):
        return getsize(self.meta_store, path)

    def listdir(self, path):
        return listdir(self.meta_store, path) class MemoryStore(Store):
    """Store class that uses a hierarchy of :class:`KVStore` objects, thus all data
    will be held in main memory.

    Examples
    --------
    This is the default class used when creating a group. E.g.::

        >>> import zarr
        >>> g = zarr.group()
        >>> type(g.store)
        <class 'zarr.storage.MemoryStore'>

    Note that the default class when creating an array is the built-in
    :class:`KVStore` class, i.e.::

        >>> z = zarr.zeros(100)
        >>> type(z.store)
        <class 'zarr.storage.KVStore'>

    Notes
    -----
    Safe to write in multiple threads.

    """

    def __init__(self, root=None, cls=dict, dimension_separator=None):
        if root is None:
            self.root = cls()
        else:
            self.root = root
        self.cls = cls
        self.write_mutex = Lock()
        self._dimension_separator = dimension_separator

    def __getstate__(self):
        return self.root, self.cls

    def __setstate__(self, state):
        root, cls = state
        self.__init__(root=root, cls=cls)

    def _get_parent(self, item: str):
        parent = self.root
        # split the item
        segments = item.split('/')
        # find the parent container
        for k in segments[:-1]:
            parent = parent[k]
            if not isinstance(parent, self.cls):
                raise KeyError(item)
        return parent, segments[-1]

    def _require_parent(self, item):
        parent = self.root
        # split the item
        segments = item.split('/')
        # require the parent container
        for k in segments[:-1]:
            try:
                parent = parent[k]
            except KeyError:
                parent[k] = self.cls()
                parent = parent[k]
            else:
                if not isinstance(parent, self.cls):
                    raise KeyError(item)
        return parent, segments[-1]

    def __getitem__(self, item: str):
        parent, key = self._get_parent(item)
        try:
            value = parent[key]
        except KeyError:
            raise KeyError(item)
        else:
            if isinstance(value, self.cls):
                raise KeyError(item)
            else:
                return value

    def __setitem__(self, item: str, value):
        with self.write_mutex:
            parent, key = self._require_parent(item)
            value = ensure_bytes(value)
            parent[key] = value

    def __delitem__(self, item: str):
        with self.write_mutex:
            parent, key = self._get_parent(item)
            try:
                del parent[key]
            except KeyError:
                raise KeyError(item)

    def __contains__(self, item: str):  # type: ignore[override]
        try:
            parent, key = self._get_parent(item)
            value = parent[key]
        except KeyError:
            return False
        else:
            return not isinstance(value, self.cls)

    def __eq__(self, other):
        return (
            isinstance(other, MemoryStore) and
            self.root == other.root and
            self.cls == other.cls
        )

    def keys(self):
        for k in _dict_store_keys(self.root, cls=self.cls):
            yield k

    def __iter__(self):
        return self.keys()

    def __len__(self) -> int:
        return sum(1 for _ in self.keys())

    def listdir(self, path: Path = None) -> List[str]:
        path = normalize_storage_path(path)
        if path:
            try:
                parent, key = self._get_parent(path)
                value = parent[key]
            except KeyError:
                return []
        else:
            value = self.root
        if isinstance(value, self.cls):
            return sorted(value.keys())
        else:
            return []

    def rename(self, src_path: Path, dst_path: Path):
        src_path = normalize_storage_path(src_path)
        dst_path = normalize_storage_path(dst_path)

        src_parent, src_key = self._get_parent(src_path)
        dst_parent, dst_key = self._require_parent(dst_path)

        dst_parent[dst_key] = src_parent.pop(src_key)

    def rmdir(self, path: Path = None):
        path = normalize_storage_path(path)
        if path:
            try:
                parent, key = self._get_parent(path)
                value = parent[key]
            except KeyError:
                return
            else:
                if isinstance(value, self.cls):
                    del parent[key]
        else:
            # clear out root
            self.root = self.cls()

    def getsize(self, path: Path = None):
        path = normalize_storage_path(path)

        # obtain value to return size of
        value = None
        if path:
            try:
                parent, key = self._get_parent(path)
                value = parent[key]
            except KeyError:
                pass
        else:
            value = self.root

        # obtain size of value
        if value is None:
            return 0

        elif isinstance(value, self.cls):
            # total size for directory
            size = 0
            for v in value.values():
                if not isinstance(v, self.cls):
                    size += buffer_size(v)
            return size

        else:
            return buffer_size(value)

    def clear(self):
        with self.write_mutex:
            self.root.clear() def atexit_rmtree(path,
                  isdir=os.path.isdir,
                  rmtree=shutil.rmtree):  # pragma: no cover
    """Ensure directory removal at interpreter exit."""
    if isdir(path):
        rmtree(path) def getsize(store: BaseStore, path: Path = None) -> int:
    """Compute size of stored items for a given path. If `store` provides a `getsize`
    method, this will be called, otherwise will return -1."""
    path = normalize_storage_path(path)
    if hasattr(store, 'getsize'):
        # pass through
        return store.getsize(path)  # type: ignore
    elif isinstance(store, MutableMapping):
        # compute from size of values
        if path in store:
            v = store[path]
            size = buffer_size(v)
        else:
            members = listdir(store, path)
            prefix = _path_to_prefix(path)
            size = 0
            for k in members:
                try:
                    v = store[prefix + k]
                except KeyError:
                    pass
                else:
                    try:
                        size += buffer_size(v)
                    except TypeError:
                        return -1
        return size
    else:
        return -1
import atexit
import tempfile
import unittest
import numpy as np
import pytest
import zarr
from numbers import Integral
from numcodecs import Adler32, Zlib
from numpy.testing import assert_array_equal
from zarr.convenience import (
    consolidate_metadata,
    copy,
    copy_store,
    load,
    open,
    open_consolidated,
    save,
    save_group,
    save_array,
    copy_all,
)
from zarr.core import Array
from zarr.errors import CopyError
from zarr.hierarchy import Group, group
from zarr.storage import (
    ConsolidatedMetadataStore,
    MemoryStore,
    atexit_rmtree,
    getsize,
)




def test_open_array(path_type):

    store = tempfile.mkdtemp()
    atexit.register(atexit_rmtree, store)
    store = path_type(store)

    # open array, create if doesn't exist
    z = open(store, mode='a', shape=100)
    assert isinstance(z, Array)
    assert z.shape == (100,)

    # open array, overwrite
    z = open(store, mode='w', shape=200)

    assert isinstance(z, Array)