Registry

Bases: Singleton

Registry for handlers

Source code in aiodistbus/registry.py
class Registry(Singleton):
    """Registry for handlers"""

    def __init__(self):
        self.namespaces: Dict[str, Namespace] = {}

    def on(
        self, event: str, dtype: Optional[Type] = None, namespace: str = "default"
    ) -> Callable:
        """Decorator to register a handler

        Args:
            event (str): Event type
            dtype (Optional[Type], optional): Data type. Defaults to None.
            namespace (str, optional): Namespace. Defaults to "default".

        Returns:
            Callable: Decorator

        Examples:
            >>> from aiodistbus import registry
            >>> @registry.on("test")
            ... def test_handler(event):
            ...     print(event)
            ...
            >>> registry.get_handlers()
            {'test': Handler(event_type='test', function=<function test_handler at 0x7f8b1c0b9d30>, dtype=None)}

        """
        # Store the handler information
        if namespace not in self.namespaces:
            self.namespaces[namespace] = Namespace()

        def decorator(func: Callable):
            # Add handler
            handler = Handler(event, func, dtype)  # <-- Missing func
            self.namespaces[namespace].handlers[event] = handler
            return func

        return decorator

    def get_handlers(self, namespace: str = "default") -> Dict[str, Handler]:
        """Get handlers for namespace

        Args:
            namespace (str, optional): Namespace. Defaults to "default".

        Returns:
            Dict[str, Handler]: Handlers

        """
        return self.namespaces[namespace].handlers

get_handlers(namespace='default')

Get handlers for namespace

Parameters:
  • namespace (str, default: 'default' ) –

    Namespace. Defaults to "default".

Returns:
  • Dict[str, Handler]

    Dict[str, Handler]: Handlers

aiodistbus/registry.py
def get_handlers(self, namespace: str = "default") -> Dict[str, Handler]:
    """Get handlers for namespace

    Args:
        namespace (str, optional): Namespace. Defaults to "default".

    Returns:
        Dict[str, Handler]: Handlers

    """
    return self.namespaces[namespace].handlers

on(event, dtype=None, namespace='default')

Decorator to register a handler

Parameters:
  • event (str) –

    Event type

  • dtype (Optional[Type], default: None ) –

    Data type. Defaults to None.

  • namespace (str, default: 'default' ) –

    Namespace. Defaults to "default".

Returns:
  • Callable( Callable ) –

    Decorator

>>> from aiodistbus import registry
>>> @registry.on("test")
... def test_handler(event):
...     print(event)
...
>>> registry.get_handlers()
{'test': Handler(event_type='test', function=<function test_handler at 0x7f8b1c0b9d30>, dtype=None)}
aiodistbus/registry.py
def on(
    self, event: str, dtype: Optional[Type] = None, namespace: str = "default"
) -> Callable:
    """Decorator to register a handler

    Args:
        event (str): Event type
        dtype (Optional[Type], optional): Data type. Defaults to None.
        namespace (str, optional): Namespace. Defaults to "default".

    Returns:
        Callable: Decorator

    Examples:
        >>> from aiodistbus import registry
        >>> @registry.on("test")
        ... def test_handler(event):
        ...     print(event)
        ...
        >>> registry.get_handlers()
        {'test': Handler(event_type='test', function=<function test_handler at 0x7f8b1c0b9d30>, dtype=None)}

    """
    # Store the handler information
    if namespace not in self.namespaces:
        self.namespaces[namespace] = Namespace()

    def decorator(func: Callable):
        # Add handler
        handler = Handler(event, func, dtype)  # <-- Missing func
        self.namespaces[namespace].handlers[event] = handler
        return func

    return decorator