zarr.registry

zarr.registry

__all__ module-attribute

__all__ = [
    "Registry",
    "get_buffer_class",
    "get_chunk_key_encoding_class",
    "get_codec_class",
    "get_ndbuffer_class",
    "get_pipeline_class",
    "register_buffer",
    "register_chunk_key_encoding",
    "register_codec",
    "register_ndbuffer",
    "register_pipeline",
]

Registry

Bases: dict[str, type[T]]

Source code in src/zarr/registry.py

class Registry[T](dict[str, type[T]]):
    def __init__(self) -> None:
        super().__init__()
        self.lazy_load_list: list[EntryPoint] = []

    def lazy_load(self, use_entrypoint_name: bool = False) -> None:
        for e in self.lazy_load_list:
            self.register(e.load(), qualname=e.name if use_entrypoint_name else None)

        self.lazy_load_list.clear()

    def register(self, cls: type[T], qualname: str | None = None) -> None:
        if qualname is None:
            qualname = fully_qualified_name(cls)
        self[qualname] = cls

lazy_load_list instance-attribute

lazy_load_list: list[EntryPoint] = []

__init__

__init__() -> None

Source code in src/zarr/registry.py

def __init__(self) -> None:
    super().__init__()
    self.lazy_load_list: list[EntryPoint] = []

lazy_load

lazy_load(use_entrypoint_name: bool = False) -> None

Source code in src/zarr/registry.py

def lazy_load(self, use_entrypoint_name: bool = False) -> None:
    for e in self.lazy_load_list:
        self.register(e.load(), qualname=e.name if use_entrypoint_name else None)

    self.lazy_load_list.clear()

register

register(
    cls: type[T], qualname: str | None = None
) -> None

Source code in src/zarr/registry.py

def register(self, cls: type[T], qualname: str | None = None) -> None:
    if qualname is None:
        qualname = fully_qualified_name(cls)
    self[qualname] = cls

fully_qualified_name

fully_qualified_name(cls: type) -> str

Source code in src/zarr/registry.py

def fully_qualified_name(cls: type) -> str:
    module = cls.__module__
    return f"{module}.{cls.__qualname__}"

get_buffer_class

get_buffer_class(
    reload_config: bool = False,
) -> type[Buffer]

Source code in src/zarr/registry.py

def get_buffer_class(reload_config: bool = False) -> type[Buffer]:
    if reload_config:
        _reload_config()
    _buffer_registry.lazy_load()

    path = config.get("buffer")
    buffer_class = _buffer_registry.get(path)
    if buffer_class:
        return buffer_class
    raise BadConfigError(
        f"Buffer class '{path}' not found in registered buffers: {list(_buffer_registry)}."
    )

get_chunk_key_encoding_class

get_chunk_key_encoding_class(
    key: str,
) -> type[ChunkKeyEncoding]

Source code in src/zarr/registry.py

def get_chunk_key_encoding_class(key: str) -> type[ChunkKeyEncoding]:
    _chunk_key_encoding_registry.lazy_load(use_entrypoint_name=True)
    if key not in _chunk_key_encoding_registry:
        raise KeyError(
            f"Chunk key encoding '{key}' not found in registered chunk key encodings: {list(_chunk_key_encoding_registry)}."
        )
    return _chunk_key_encoding_registry[key]

get_codec_class

get_codec_class(
    key: str, reload_config: bool = False
) -> type[Codec]

Source code in src/zarr/registry.py

def get_codec_class(key: str, reload_config: bool = False) -> type[Codec]:
    if reload_config:
        _reload_config()

    if key in _codec_registries:
        # logger.debug("Auto loading codec '%s' from entrypoint", codec_id)
        _codec_registries[key].lazy_load()

    codec_classes = _codec_registries[key]
    if not codec_classes:
        raise UnknownCodecError(_missing_codec_message(key, zarr_format=3))
    config_entry = config.get("codecs", {}).get(key)
    if config_entry is None:
        if len(codec_classes) == 1:
            return next(iter(codec_classes.values()))
        warnings.warn(
            f"Codec '{key}' not configured in config. Selecting any implementation.",
            stacklevel=2,
            category=ZarrUserWarning,
        )
        return list(codec_classes.values())[-1]
    selected_codec_cls = codec_classes.get(config_entry)
    if selected_codec_cls is None:
        # Not UnknownCodecError: the codec is known, the implementation named in the config is
        # not registered. That is a configuration problem, which is what the sibling getters in
        # this module raise BadConfigError for.
        raise BadConfigError(
            f"Codec {key!r} is configured to use the implementation {config_entry!r}, which is "
            f"not registered. Registered implementations of this codec: "
            f"{sorted(codec_classes)}."
        )
    return selected_codec_cls

get_ndbuffer_class

get_ndbuffer_class(
    reload_config: bool = False,
) -> type[NDBuffer]

Source code in src/zarr/registry.py

def get_ndbuffer_class(reload_config: bool = False) -> type[NDBuffer]:
    if reload_config:
        _reload_config()
    _ndbuffer_registry.lazy_load()
    path = config.get("ndbuffer")
    ndbuffer_class = _ndbuffer_registry.get(path)
    if ndbuffer_class:
        return ndbuffer_class
    raise BadConfigError(
        f"NDBuffer class '{path}' not found in registered buffers: {list(_ndbuffer_registry)}."
    )

get_numcodec

get_numcodec(data: CodecJSON_V2[str]) -> Numcodec

Resolve a numcodec codec from the numcodecs registry.

This requires the Numcodecs package to be installed.

Parameters:

• data (CodecJSON_V2) –

  The JSON metadata for the codec.

Returns:

• codec ( Numcodec ) –

Raises:

• UnknownCodecError –

  If data carries a string "id" that is not registered with numcodecs. Any other failure, including a registered codec rejecting its configuration and a data that is not a mapping, propagates from numcodecs unchanged.

Examples:

from zarr.registry import get_numcodec
codec = get_numcodec({'id': 'zlib', 'level': 1})
codec
# Zlib(level=1)

Source code in src/zarr/registry.py

def get_numcodec(data: CodecJSON_V2[str]) -> Numcodec:
    """
    Resolve a numcodec codec from the numcodecs registry.

    This requires the Numcodecs package to be installed.

    Parameters
    ----------
    data : CodecJSON_V2
        The JSON metadata for the codec.

    Returns
    -------
    codec : Numcodec

    Raises
    ------
    UnknownCodecError
        If ``data`` carries a string ``"id"`` that is not registered with numcodecs. Any other
        failure, including a registered codec rejecting its configuration and a ``data`` that is
        not a mapping, propagates from numcodecs unchanged.

    Examples
    --------
    ```python
    from zarr.registry import get_numcodec
    codec = get_numcodec({'id': 'zlib', 'level': 1})
    codec
    # Zlib(level=1)
    ```
    """

    from numcodecs.registry import codec_registry, entries, get_codec

    # Check whether numcodecs can resolve the id *before* handing off, rather than catching what
    # `get_codec` raises. Catching cannot tell "this id is unregistered" from "a registered codec
    # rejected its configuration" or from "a wrapper codec failed to resolve an inner codec", and
    # relabelling either of those with this id would attach a package hint that is simply wrong.
    # This mirrors the two lookups `get_codec` performs (it then tests the result for
    # truthiness rather than membership, which only differs for a falsy registry value).
    # Widened to `object` deliberately: `data` is annotated as a TypedDict, but this is a public
    # function and callers pass whatever they like. numcodecs coerces with `dict(config)` and
    # raises for anything that is not a mapping, which is the behaviour to preserve.
    raw: object = data
    codec_id = raw.get("id") if isinstance(raw, Mapping) else None
    if isinstance(codec_id, str) and codec_id not in codec_registry and codec_id not in entries:
        raise UnknownCodecError(_missing_codec_message(codec_id, zarr_format=2))
    return get_codec(data)  # type: ignore[no-any-return]

get_pipeline_class

get_pipeline_class(
    reload_config: bool = False,
) -> type[CodecPipeline]

Source code in src/zarr/registry.py

def get_pipeline_class(reload_config: bool = False) -> type[CodecPipeline]:
    if reload_config:
        _reload_config()
    _pipeline_registry.lazy_load()
    path = config.get("codec_pipeline.path")
    pipeline_class = _pipeline_registry.get(path)
    if pipeline_class:
        return pipeline_class
    raise BadConfigError(
        f"Pipeline class '{path}' not found in registered pipelines: {list(_pipeline_registry)}."
    )

register_buffer

register_buffer(
    cls: type[Buffer], qualname: str | None = None
) -> None

Source code in src/zarr/registry.py

def register_buffer(cls: type[Buffer], qualname: str | None = None) -> None:
    _buffer_registry.register(cls, qualname)

register_chunk_key_encoding

register_chunk_key_encoding(key: str, cls: type) -> None

Source code in src/zarr/registry.py

def register_chunk_key_encoding(key: str, cls: type) -> None:
    _chunk_key_encoding_registry.register(cls, key)

register_codec

register_codec(
    key: str,
    codec_cls: type[Codec],
    *,
    qualname: str | None = None,
) -> None

Source code in src/zarr/registry.py

def register_codec(key: str, codec_cls: type[Codec], *, qualname: str | None = None) -> None:
    if key not in _codec_registries:
        _codec_registries[key] = Registry()
    _codec_registries[key].register(codec_cls, qualname=qualname)

register_ndbuffer

register_ndbuffer(
    cls: type[NDBuffer], qualname: str | None = None
) -> None

Source code in src/zarr/registry.py

def register_ndbuffer(cls: type[NDBuffer], qualname: str | None = None) -> None:
    _ndbuffer_registry.register(cls, qualname)

register_pipeline

register_pipeline(pipe_cls: type[CodecPipeline]) -> None

Source code in src/zarr/registry.py

def register_pipeline(pipe_cls: type[CodecPipeline]) -> None:
    _pipeline_registry.register(pipe_cls)