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: |
|
|---|
| Returns: |
|
|---|
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: |
|
|---|
| Returns: |
|
|---|
>>> 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