dddmisc-handlers-collection

d3m.hc

HandlersCollection

Bases: IHandlersCollection

Class representing a collection of command and event handlers.

The HandlersCollection class provides the following functionality

• Registration and retrieval of command handlers
• Retrieval of event handlers
• Registration of event subscriptions

Parameters:

Name	Type	Description	Default
name	str	The name of the collection	None

Methods:

Name	Description
get_command_handler	Retrieves the command handler for the specified command.
get_event_handlers	Retrieves the event handlers for the specified event.
register	Registers a command handler.
subscribe	Registers an event subscription.
set_defaults	Sets default values for the specified domain.
__copy__	Creates and returns a copy of the HandlersCollection object.

Examples:

Init collection:

>>> collection = HandlersCollection(name='Example')
>>> collection
d3m.hc.HandlersCollection<name='Example', registered_command=0, registered_events=0>

Register command's handler:

>>> @collection.register
... async def command_handler1(cmd: CustomCommand1, dependencies1: int):
...     ...
>>> @collection.register
... async def command_handler2(dependencies1: int, cmd: CustomCommand2):
...     ...
>>>
>>> collection
d3m.hc.HandlersCollection<name='Example', registered_command=2, registered_events=0>

Handlers collection associate command with command's argument by signature type hinting!

Subscribe to event:

>>> @collection.subscribe('other-domain.CustomEvent')
... @collection.register
... async def command_handler(arg1: str, cmd: CustomCommand):
...     return cmd.__reference__
>>> collection
d3m.hc.HandlersCollection<name='Example', registered_command=1, registered_events=1>

get_command_handler(__command, **dependencies)

Retrieves the command handler for the specified command.

Parameters:

Name	Type	Description	Default
__command	AbstractCommand	The command to be handled.	required
**dependencies		Additional dependencies that the command handler may require.	{}

Returns:

Type	Description
Callable[[], Coroutine]	A callable that is a coroutine and represents the handler for the given command.

Examples:

>>> collection = HandlersCollection(name='Example')
>>> @collection.register
... async def command_handler(arg1: str, cmd: CustomCommand):
...     return cmd.__reference__, arg1
>>> cmd = CustomCommand()
>>> handler = collection.get_command_handler(cmd, arg1='abc')
>>> assert await handler() == (cmd.__reference__, 'abc')

get_event_handlers(__event, /, **dependencies)

Retrieves the event handlers for the specified event.

Parameters:

Name	Type	Description	Default
__event	AbstractEvent	An instance of AbstractEvent representing the event.	required
**dependencies		Additional dependencies required by the event handlers.	{}

Returns:

Type	Description
tuple[Callable[[], Coroutine], ...]	A tuple of callables that represent the event handlers for the given event. Each callable is a coroutine that takes no arguments.

Examples:

>>> collection = HandlersCollection(name='Example')
>>> @collection.subscribe('other-domain.CustomEvent')
>>> @collection.register
... async def command_handler1(arg1: str, cmd: CustomCommand1):
...     return arg1
>>> @collection.subscribe('other-domain.CustomEvent')
>>> @collection.register
... async def command_handler2(arg2: str, cmd: CustomCommand2):
...     return arg2
>>> event = UniversalMessage('other-domain.CustomEvent', 'event', {})
>>> handlers = collection.get_event_handlers(event, arg1='abc', arg2='xyz')
>>> assert len(handlers) == 2
>>> assert await handlers[0]() == 'abc'
>>> assert await handlers[1]() == 'xyz'

register(func)

Registers a command handler.

Parameters:

Name	Type	Description	Default
func	Callable[..., Coroutine]	A callable object that takes any number of arguments and returns a coroutine.	required

Returns:

Type	Description
ICommandHandler	An instance of ICommandHandler representing the registered handler.

set_defaults(__domain, /, **defaults)

Sets default values for handler's dependencies. Dependencies associated with handlers by domain and attribute name.

Parameters:

Name	Type	Description	Default
__domain	str | DomainName	The domain for which default values are being set. Can be either a string or a DomainName object.	required
**defaults		Dependencies values for the specified domain.	{}

subscribe(full_event_name, /, *, condition=none_condition, converter=lambda payload: payload, retry=tc.retry_never, stop=tc.stop_after_attempt(1), wait=tc.wait_none())

Registers an event subscription.

Parameters:

Name	Type	Description	Default
full_event_name	str	A string representing the full name of the event.	required
condition	ICondition	An instance of ICondition. (Default: none_condition)	none_condition
converter	Callable[[Mapping], Mapping]	A Callable object using for conversion event's payload to command's payload. Can used as Anti-coraption layer between events from other domain. (Default: lambda payload: payload)	lambda payload: payload
retry	retry_base	retry exec event's handler strategy. (Default: tenacity.retry_never)	retry_never
stop	stop_base	stop retry strategy. (Default: tenacity.stop_after_attempt(1))	stop_after_attempt(1)
wait	wait_base	wait between attempts strategy. (Default: tenacity.wait_none())	wait_none()

Returns:

Type	Description
Callable[[ICommandHandler], ICommandHandler]	A ICommandHandler's decorator. Not change handler behaviour.

d3m.hc.conditions

ICondition

Bases: ABC

Base interface for conditions classes used for filter events

Methods:

Name	Description
check	Checks the event against the specified condition

Returns:

Name	Type	Description
bool		True when consistent, False when inconsistent

check(event) abstractmethod

Checks the event against the specified condition

Parameters:

Name	Type	Description	Default
event	AbstractEvent	event	required

Returns:

Type	Description
bool	True when consistent, False when inconsistent

HasAttrs

Bases: ICondition

A condition that checks if an event has all the specified attributes.

Parameters:

Name	Type	Description	Default
*attrs	str	A variable number of attribute names as strings.	()

Methods:

Name	Description
check	Checks if an event has all the specified attributes.

Examples:

>>> has_attrs = HasAttrs('attribute1', 'attribute2')
>>> event = MyEvent({'attribute1': 'value1', 'attribute2': 'value2', 'attribute3': 'value3'})
>>> has_attrs.check(event)
True

And

Bases: ICondition

"And" condition.

A class representing a logical AND condition.

Parameters:

Name	Type	Description	Default
*conditions	ICondition	Initializes an And instance with the given conditions.	()

Methods:

Name	Description
check	Checks if an event against all the specified conditions

Or

Bases: ICondition

"Or" condition.

A class representing a logical OR condition.

Parameters:

Name	Type	Description	Default
*conditions	ICondition	Initializes an Or instance with the given conditions.	()

Methods:

Name	Description
check	Checks if an event against any the specified conditions

Not

Bases: ICondition

This class represents a logical NOT condition that can be used to negate the result of another condition.

Parameters:

Name	Type	Description	Default
condition	ICondition	The condition to be negated.	required

Methods:

Name	Description
check	Checks if an event not against the specified condition

Equal

Bases: ICondition

Class representing an 'Equal' condition.

This class implements the 'ICondition' interface and provides a way to check if certain attributes in an event's payload are equal to specified values.

Parameters:

Name	Type	Description	Default
**attrs		A containing the attributes and values to check.	{}

Methods:

Name	Description
check	Checks if the attributes in the event's payload are equal to the specified values.