RegistryManager

A class to manage multiple named registries. NOTE: Ensure that the register decorator is the outermost decorator!

Source code in wt_ml/tasklib.py
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
class RegistryManager:
    """
    A class to manage multiple named registries.
    NOTE: Ensure that the register decorator is the outermost decorator!
    """

    _registries: dict[str, dict[str, Callable]] = {}
    _registers: dict[str, Callable] = {}

    @staticmethod
    def create_registry(name: str) -> tuple[Callable, dict[str, Callable]]:
        """
        Create a new named registry.
        NOTE: The register decorator returned must be the outermost decorator as it might not work correctly.

        Args:
            name (str): A unique name for the registry.

        Returns:
            tuple: A tuple containing the decorator function and the registry dictionary

        Raises:
            ValueError: If a registry with the given name already exists.
        """
        if name in RegistryManager._registries:
            raise ValueError(f"Registry '{name}' already exists")

        registry = {}

        # TODO (@legendof-selda): Find a way to check if the decorator used is the outermost decorator
        def register(node: Callable, rename: str | None = None) -> Callable:
            """
            Decorator function to register functions in the registry.

            Args:
                node (Callable): Node function to be registered.
                rename (str | None): Rename the given node. Defaults to None.

            Returns:
                Callable: The original function

            Raises:
                KeyError: If a function with the same name already exists in the registry.
            """
            key = rename or node.__name__

            if key in registry:
                raise KeyError(f"Key '{key}' already exists in registry '{name}'")

            if not is_outermost_decorator(node):
                logger.warning(f"@register is not the outermost decorator for func `{key}`. This can lead to issues.")

            registry[key] = node

            return node

        register.registry = registry
        RegistryManager._registries[name] = registry
        RegistryManager._registers[name] = register

        return register, registry

    @staticmethod
    def get_registry(name: str) -> dict[str, Callable]:
        """
        Retrieve a specific registry by name.

        Args:
            name (str): The name of the registry to retrieve.

        Returns:
            dict: The registry dictionary.

        Raises:
            KeyError: If the registry does not exist.
        """
        return RegistryManager._registries[name]

    @staticmethod
    def get_registrar(name: str) -> Callable:
        """
        Retrieve a specific register decorator by name.

        Args:
            name (str): The name of the register decorator to retrieve.

        Returns:
            dict: The register decorator.

        Raises:
            KeyError: If the register decorator does not exist.
        """
        return RegistryManager._registers[name]

    @staticmethod
    def list_registries() -> list[str]:
        """
        List all existing registry names.

        Returns:
            list: Names of all created registries.
        """
        return list(RegistryManager._registries.keys())

create_registry(name) staticmethod

Create a new named registry. NOTE: The register decorator returned must be the outermost decorator as it might not work correctly.

Parameters:

Name Type Description Default
name str

A unique name for the registry.

required

Returns:

Name Type Description
tuple tuple[Callable, dict[str, Callable]]

A tuple containing the decorator function and the registry dictionary

Raises:

Type Description
ValueError

If a registry with the given name already exists.

Source code in wt_ml/tasklib.py
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
@staticmethod
def create_registry(name: str) -> tuple[Callable, dict[str, Callable]]:
    """
    Create a new named registry.
    NOTE: The register decorator returned must be the outermost decorator as it might not work correctly.

    Args:
        name (str): A unique name for the registry.

    Returns:
        tuple: A tuple containing the decorator function and the registry dictionary

    Raises:
        ValueError: If a registry with the given name already exists.
    """
    if name in RegistryManager._registries:
        raise ValueError(f"Registry '{name}' already exists")

    registry = {}

    # TODO (@legendof-selda): Find a way to check if the decorator used is the outermost decorator
    def register(node: Callable, rename: str | None = None) -> Callable:
        """
        Decorator function to register functions in the registry.

        Args:
            node (Callable): Node function to be registered.
            rename (str | None): Rename the given node. Defaults to None.

        Returns:
            Callable: The original function

        Raises:
            KeyError: If a function with the same name already exists in the registry.
        """
        key = rename or node.__name__

        if key in registry:
            raise KeyError(f"Key '{key}' already exists in registry '{name}'")

        if not is_outermost_decorator(node):
            logger.warning(f"@register is not the outermost decorator for func `{key}`. This can lead to issues.")

        registry[key] = node

        return node

    register.registry = registry
    RegistryManager._registries[name] = registry
    RegistryManager._registers[name] = register

    return register, registry

get_registrar(name) staticmethod

Retrieve a specific register decorator by name.

Parameters:

Name Type Description Default
name str

The name of the register decorator to retrieve.

required

Returns:

Name Type Description
dict Callable

The register decorator.

Raises:

Type Description
KeyError

If the register decorator does not exist.

Source code in wt_ml/tasklib.py
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
@staticmethod
def get_registrar(name: str) -> Callable:
    """
    Retrieve a specific register decorator by name.

    Args:
        name (str): The name of the register decorator to retrieve.

    Returns:
        dict: The register decorator.

    Raises:
        KeyError: If the register decorator does not exist.
    """
    return RegistryManager._registers[name]

get_registry(name) staticmethod

Retrieve a specific registry by name.

Parameters:

Name Type Description Default
name str

The name of the registry to retrieve.

required

Returns:

Name Type Description
dict dict[str, Callable]

The registry dictionary.

Raises:

Type Description
KeyError

If the registry does not exist.

Source code in wt_ml/tasklib.py
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
@staticmethod
def get_registry(name: str) -> dict[str, Callable]:
    """
    Retrieve a specific registry by name.

    Args:
        name (str): The name of the registry to retrieve.

    Returns:
        dict: The registry dictionary.

    Raises:
        KeyError: If the registry does not exist.
    """
    return RegistryManager._registries[name]

list_registries() staticmethod

List all existing registry names.

Returns:

Name Type Description
list list[str]

Names of all created registries.

Source code in wt_ml/tasklib.py
619
620
621
622
623
624
625
626
627
@staticmethod
def list_registries() -> list[str]:
    """
    List all existing registry names.

    Returns:
        list: Names of all created registries.
    """
    return list(RegistryManager._registries.keys())