confit.registry

Registry

Bases: catalogue.Registry

A registry that validates the input arguments of the registered functions.

Source code in confit/registry.py

class Registry(catalogue.Registry):
    """
    A registry that validates the input arguments of the registered functions.
    """

    def __init__(self, namespace: Sequence[str], entry_points: bool = False) -> None:
        """
        Initialize the registry.

        Parameters
        ----------
        namespace: Sequence[str]
            The namespace of the registry
        entry_points: bool
            Should we use entry points to load the registered functions
        """
        super().__init__(namespace, entry_points=entry_points)
        self.registry = None

    def register(
        self,
        name: str,
        *,
        func: Optional[catalogue.InFunc] = None,
        save_params: Optional[Dict[str, Any]] = None,
        skip_save_params: Sequence[str] = (),
        invoker: Optional[Callable] = None,
    ) -> Callable[[catalogue.InFunc], catalogue.InFunc]:
        """
        This is a convenience wrapper around `catalogue.Registry.register`, that
        additionally validates the input arguments of the registered function and
        saves the result of any call to a mapping to its arguments.

        Parameters
        ----------
        name: str
            The name of the function
        func: Optional[catalogue.InFunc]
            The function to register
        save_params: Optional[Dict[str, Any]]
            Additional parameters to save when the function is called. If falsy,
            the function parameters are not saved
        skip_save_params: Sequence[str]
            List of parameters to skip when saving the function parameters
        invoker: Optional[Callable] = None,
            An optional invoker to apply to the function before registering it.
            It is better to use this than to apply the invoker to the function
            to preserve the signature of the function or the class and enable
            validating its parameters.

        Returns
        -------
        Callable[[catalogue.InFunc], catalogue.InFunc]
        """
        registerer = super().register(name)

        save_params = save_params or {f"@{self.namespace[-1]}": name}

        def invoke(func, params):
            resolved = invoker(func, params) if invoker is not None else func(params)
            if save_params is not None:
                params_to_save = {**save_params, **params}
                for name in skip_save_params:
                    params_to_save.pop(name, None)
                Config._store_resolved(resolved, params_to_save)
            return resolved

        def wrap_and_register(fn: catalogue.InFunc) -> catalogue.InFunc:

            if save_params is not None:
                _check_signature_for_save_params(
                    fn if not isinstance(fn, type) else fn.__init__
                )

            validated_fn = validate_arguments(
                fn,
                config={"arbitrary_types_allowed": True},
                registry=getattr(self, "registry", None),
                invoker=invoke,
            )
            registerer(validated_fn)
            return validated_fn

        if func is not None:
            return wrap_and_register(func)
        else:
            return wrap_and_register

    def get_available(self) -> Sequence[str]:
        """Get all functions for a given namespace.

        namespace (Tuple[str]): The namespace to get.
        RETURNS (Dict[str, Any]): The functions, keyed by name.
        """
        result = set()
        if self.entry_points:
            result.update({p.name for p in self._get_entry_points()})
        for keys in catalogue.REGISTRY.copy().keys():
            if len(self.namespace) == len(keys) - 1 and all(
                self.namespace[i] == keys[i] for i in range(len(self.namespace))
            ):
                result.add(keys[-1])
        return sorted(result)

__init__(namespace, entry_points=False)

Initialize the registry.

PARAMETER	DESCRIPTION
namespace	The namespace of the registry TYPE: Sequence[str]
entry_points	Should we use entry points to load the registered functions TYPE: bool DEFAULT: False

Source code in confit/registry.py

def __init__(self, namespace: Sequence[str], entry_points: bool = False) -> None:
    """
    Initialize the registry.

    Parameters
    ----------
    namespace: Sequence[str]
        The namespace of the registry
    entry_points: bool
        Should we use entry points to load the registered functions
    """
    super().__init__(namespace, entry_points=entry_points)
    self.registry = None

register(name, *, func=None, save_params=None, skip_save_params=(), invoker=None)

This is a convenience wrapper around catalogue.Registry.register, that additionally validates the input arguments of the registered function and saves the result of any call to a mapping to its arguments.

PARAMETER	DESCRIPTION
name	The name of the function TYPE: str
func	The function to register TYPE: Optional[catalogue.InFunc] DEFAULT: None
save_params	Additional parameters to save when the function is called. If falsy, the function parameters are not saved TYPE: Optional[Dict[str, Any]] DEFAULT: None
skip_save_params	List of parameters to skip when saving the function parameters TYPE: Sequence[str] DEFAULT: ()
invoker	An optional invoker to apply to the function before registering it. It is better to use this than to apply the invoker to the function to preserve the signature of the function or the class and enable validating its parameters. TYPE: Optional[Callable] DEFAULT: None

RETURNS	DESCRIPTION
Callable[[catalogue.InFunc], catalogue.InFunc]

Source code in confit/registry.py

def register(
    self,
    name: str,
    *,
    func: Optional[catalogue.InFunc] = None,
    save_params: Optional[Dict[str, Any]] = None,
    skip_save_params: Sequence[str] = (),
    invoker: Optional[Callable] = None,
) -> Callable[[catalogue.InFunc], catalogue.InFunc]:
    """
    This is a convenience wrapper around `catalogue.Registry.register`, that
    additionally validates the input arguments of the registered function and
    saves the result of any call to a mapping to its arguments.

    Parameters
    ----------
    name: str
        The name of the function
    func: Optional[catalogue.InFunc]
        The function to register
    save_params: Optional[Dict[str, Any]]
        Additional parameters to save when the function is called. If falsy,
        the function parameters are not saved
    skip_save_params: Sequence[str]
        List of parameters to skip when saving the function parameters
    invoker: Optional[Callable] = None,
        An optional invoker to apply to the function before registering it.
        It is better to use this than to apply the invoker to the function
        to preserve the signature of the function or the class and enable
        validating its parameters.

    Returns
    -------
    Callable[[catalogue.InFunc], catalogue.InFunc]
    """
    registerer = super().register(name)

    save_params = save_params or {f"@{self.namespace[-1]}": name}

    def invoke(func, params):
        resolved = invoker(func, params) if invoker is not None else func(params)
        if save_params is not None:
            params_to_save = {**save_params, **params}
            for name in skip_save_params:
                params_to_save.pop(name, None)
            Config._store_resolved(resolved, params_to_save)
        return resolved

    def wrap_and_register(fn: catalogue.InFunc) -> catalogue.InFunc:

        if save_params is not None:
            _check_signature_for_save_params(
                fn if not isinstance(fn, type) else fn.__init__
            )

        validated_fn = validate_arguments(
            fn,
            config={"arbitrary_types_allowed": True},
            registry=getattr(self, "registry", None),
            invoker=invoke,
        )
        registerer(validated_fn)
        return validated_fn

    if func is not None:
        return wrap_and_register(func)
    else:
        return wrap_and_register

get_available()

Get all functions for a given namespace.

namespace (Tuple[str]): The namespace to get. RETURNS (Dict[str, Any]): The functions, keyed by name.

Source code in confit/registry.py

def get_available(self) -> Sequence[str]:
    """Get all functions for a given namespace.

    namespace (Tuple[str]): The namespace to get.
    RETURNS (Dict[str, Any]): The functions, keyed by name.
    """
    result = set()
    if self.entry_points:
        result.update({p.name for p in self._get_entry_points()})
    for keys in catalogue.REGISTRY.copy().keys():
        if len(self.namespace) == len(keys) - 1 and all(
            self.namespace[i] == keys[i] for i in range(len(self.namespace))
        ):
            result.add(keys[-1])
    return sorted(result)

MetaRegistryCollection

Bases: type

A metaclass for the registry collection that adds it as the registry collection of all registries defined in the body of the class.

Source code in confit/registry.py

class MetaRegistryCollection(type):
    """
    A metaclass for the registry collection that adds it as the
    registry collection of all registries defined in the body of the class.
    """

    def __setattr__(self, key, value):
        assert isinstance(value, Registry)
        value.registry = self
        super().__setattr__(key, value)

    def __init__(cls, name, bases, dct):
        """
        Initialize the registry collection by adding it-self as the registry collection
        of all registries.

        Parameters
        ----------
        name
        bases
        dct
        """
        super().__init__(name, bases, dct)
        for key, value in dct.items():
            if isinstance(value, Registry):
                value.registry = cls

__init__(name, bases, dct)

Initialize the registry collection by adding it-self as the registry collection of all registries.

PARAMETER	DESCRIPTION
name
bases
dct

Source code in confit/registry.py

def __init__(cls, name, bases, dct):
    """
    Initialize the registry collection by adding it-self as the registry collection
    of all registries.

    Parameters
    ----------
    name
    bases
    dct
    """
    super().__init__(name, bases, dct)
    for key, value in dct.items():
        if isinstance(value, Registry):
            value.registry = cls

RegistryCollection

A collection of registries.

```python class MyRegistries(RegistryCollection): my_registry = Registry(("package_name", "my_registry"), entry_points=True) my_other_registry = Registry(("package_name", "my_other_registry"))

Source code in confit/registry.py

class RegistryCollection(metaclass=MetaRegistryCollection):
    """
    A collection of registries.

    ```python
    class MyRegistries(RegistryCollection):
        my_registry = Registry(("package_name", "my_registry"), entry_points=True)
        my_other_registry = Registry(("package_name", "my_other_registry"))
    """

validate_arguments(func=None, *, config=None, invoker=None, registry=None)

Decorator to validate the arguments passed to a function and store the result in a mapping from results to call parameters (allowing

PARAMETER	DESCRIPTION
func	The function or class to call TYPE: Optional[Callable] DEFAULT: None
config	The validation configuration object TYPE: Dict DEFAULT: None
invoker	An optional invoker to apply on the validated function TYPE: Optional[Callable[[Callable, Dict[str, Any]], Any]] DEFAULT: None
registry	The registry to use to resolve the default parameters TYPE: Any DEFAULT: None

RETURNS	DESCRIPTION
Any

Source code in confit/registry.py

def validate_arguments(
    func: Optional[Callable] = None,
    *,
    config: Dict = None,
    invoker: Optional[Callable[[Callable, Dict[str, Any]], Any]] = None,
    registry: Any = None,
) -> Any:
    """
    Decorator to validate the arguments passed to a function and store the result
    in a mapping from results to call parameters (allowing

    Parameters
    ----------
    func: Callable
        The function or class to call
    config: Dict
        The validation configuration object
    invoker: Optional[Callable]
        An optional invoker to apply on the validated function
    registry: Any
        The registry to use to resolve the default parameters

    Returns
    -------
    Any
    """
    if config is None:
        config = {}
    config = {**config, "arbitrary_types_allowed": True}

    def validate(_func: Callable) -> Callable:
        if isinstance(_func, type):
            _func: type
            if hasattr(_func.__init__, "raw_function"):
                vd = pydantic.decorator.ValidatedFunction(
                    _func.__init__.raw_function, config
                )
            else:
                vd = pydantic.decorator.ValidatedFunction(_func.__init__, config)
            vd.model.__name__ = _func.__name__
            vd.model.__fields__["self"].required = False

            # Should we store the generator instead ?
            existing_validators = (
                list(_func.__get_validators__())
                if hasattr(_func, "__get_validators__")
                else []
            )

            # This function is called by Pydantic when asked to cast
            # a value (most likely a dict) as a Model (most often during
            # a function call)
            def __get_validators__():
                """
                This function is called by Pydantic when asked to cast
                a value (most likely a dict) as a Model (most often during
                a function call)

                Yields
                -------
                Callable
                    The validator function
                """

                def _validate(value):
                    if isinstance(value, dict):
                        value = Config(value).resolve(registry=registry)

                    for validator in existing_validators:
                        value = validator(value)

                    if isinstance(value, _func):
                        return value

                    return _func(**value)

                yield _validate

            # This function is called when we do Model(variable=..., other=...)
            @wraps(vd.raw_function)
            def wrapper_function(*args: Any, **kwargs: Any) -> Any:
                return _resolve_and_validate_call(
                    args=args,
                    kwargs=kwargs,
                    pydantic_func=vd,
                    use_self=True,
                    invoker=invoker,
                )

            _func.vd = vd  # type: ignore
            _func.__get_validators__ = __get_validators__  # type: ignore
            _func.model = vd.model  # type: ignore
            _func.__init__ = wrapper_function
            _func.__init__.raw_function = vd.raw_function  # type: ignore
            return _func

        else:
            vd = pydantic.decorator.ValidatedFunction(_func, config)

            @wraps(_func)
            def wrapper_function(*args: Any, **kwargs: Any) -> Any:
                return _resolve_and_validate_call(
                    args=args,
                    kwargs=kwargs,
                    pydantic_func=vd,
                    use_self=False,
                    invoker=invoker,
                )

            wrapper_function.vd = vd  # type: ignore
            wrapper_function.validate = vd.init_model_instance  # type: ignore
            wrapper_function.raw_function = vd.raw_function  # type: ignore
            wrapper_function.model = vd.model  # type: ignore
            return wrapper_function

    if func:
        return validate(func)
    else:
        return validate

get_default_registry()

Get the default registered registry.

RETURNS	DESCRIPTION
Registry

Source code in confit/registry.py

def get_default_registry() -> Any:
    """
    Get the default registered registry.

    Returns
    -------
    Registry
    """
    return _default_registry

set_default_registry(registry)

Set the default registered registry. This is used in Config.resolve() when no registry is provided.

PARAMETER	DESCRIPTION
registry	TYPE: CustomRegistry

RETURNS	DESCRIPTION
Registry

Source code in confit/registry.py

def set_default_registry(registry: CustomRegistry) -> CustomRegistry:
    """
    Set the default registered registry. This is used in
    [`Config.resolve()`][confit.config.Config.resolve] when no registry is provided.

    Parameters
    ----------
    registry: Registry

    Returns
    -------
    Registry
    """
    global _default_registry
    _default_registry = registry
    return registry