/ludwig/utils/registry.py

#! /usr/bin/env python
# Copyright (c) 2023 Predibase, Inc., 2020 Uber Technologies, Inc.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#     http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
# ==============================================================================

from collections import UserDict
from typing import Generic, TypeVar

# Legacy default keys for backward compatibility.
# New code should use the explicit `default` parameter instead.
DEFAULT_KEYS = ["None", "none", "null", None]


T = TypeVar("T")


class Registry(UserDict, Generic[T]):
    """Type-safe registry with optional parent delegation and mock support.

    Items are considered to exist in the registry if they are in either the
    registry itself or its parent. Supports:
    - Generic typing: Registry[EncoderType], Registry[CombinerType]
    - Parent delegation for hierarchical registries
    - register() decorator for clean registration
    - unregister() for testing and dynamic removal
    - Mock support via context manager
    """

    def __init__(self, source=None):
        init_data = None
        parent = {}
        if isinstance(source, Registry):
            parent = source
        else:
            init_data = source

        self.parent = parent
        super().__init__(init_data)

    def __getitem__(self, key: str) -> T:
        if self.parent and key not in self.data:
            return self.parent.__getitem__(key)
        return self.data.__getitem__(key)

    def __contains__(self, key: str):
        return key in self.data or key in self.parent

    def __len__(self) -> int:
        return len(self.data) + len(self.parent)

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

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

    def values(self):
        return self._merged().values()

    def items(self):
        return self._merged().items()

    def _merged(self):
        return {**self.parent, **self.data}

    def register(self, name: str, default: bool = False):
        """Register a class in the registry via decorator.

        Args:
            name: Registration key.
            default: If True, also register under None/"none"/"null" keys.
        """

        def wrap(cls):
            self[name] = cls
            if default:
                for key in DEFAULT_KEYS:
                    self[key] = cls
            return cls

        return wrap

    def unregister(self, name: str):
        """Remove a registered item. Useful for testing.

        Args:
            name: Key to remove.

        Raises:
            KeyError if name is not registered.
        """
        if name in self.data:
            del self.data[name]
        else:
            raise KeyError(f"'{name}' is not registered")

    def get_default(self) -> T | None:
        """Get the default registered item (registered with default=True)."""
        for key in DEFAULT_KEYS:
            if key in self.data:
                return self.data[key]
        return None

    def list_registered(self) -> list[str]:
        """List all registered names (excluding default key aliases)."""
        return [k for k in self._merged() if k not in DEFAULT_KEYS]