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 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 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 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 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 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 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 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 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 KeyError(key)
    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[config_entry]

    if selected_codec_cls:
        return selected_codec_cls
    raise KeyError(key)

get_ndbuffer_class

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

Source code in 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 ) –

Examples:

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

Source code in 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

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

    from numcodecs.registry import get_codec

    return get_codec(data)  # type: ignore[no-any-return]

get_pipeline_class

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

Source code in 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 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 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 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 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 zarr/registry.py

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