registry

zarr.registry

T module-attribute

T = TypeVar('T')

__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",
]

__buffer_registry module-attribute

__buffer_registry: Registry[Buffer] = Registry()

__chunk_grid_registry module-attribute

__chunk_grid_registry: Registry[ChunkGrid] = Registry()

__chunk_key_encoding_registry module-attribute

__chunk_key_encoding_registry: Registry[
    ChunkKeyEncoding
] = Registry()

The registry module is responsible for managing implementations of codecs, pipelines, buffers, ndbuffers, and chunk key encodings and collecting them from entrypoints. The implementation used is determined by the config.

The registry module is also responsible for managing dtypes.

__codec_registries module-attribute

__codec_registries: dict[str, Registry[Codec]] = (
    defaultdict(Registry)
)

__ndbuffer_registry module-attribute

__ndbuffer_registry: Registry[NDBuffer] = Registry()

__pipeline_registry module-attribute

__pipeline_registry: Registry[CodecPipeline] = Registry()

Registry

Bases: dict[str, type[T]], Generic[T]

Source code in zarr/registry.py

class Registry(dict[str, type[T]], Generic[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 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_grid_class

get_chunk_grid_class(key: str) -> type[ChunkGrid]

Source code in zarr/registry.py

def get_chunk_grid_class(key: str) -> type[ChunkGrid]:
    __chunk_grid_registry.lazy_load(use_entrypoint_name=True)
    if key not in __chunk_grid_registry:
        raise KeyError(
            f"Chunk grid '{key}' not found in registered chunk grids: "
            f"{list(__chunk_grid_registry)}."
        )
    return __chunk_grid_registry[key]

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_grid

register_chunk_grid(key: str, cls: type[ChunkGrid]) -> None

Source code in zarr/registry.py

def register_chunk_grid(key: str, cls: type[ChunkGrid]) -> None:
    __chunk_grid_registry.register(cls, key)

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)