Dispatchers

class domprob.dispatchers.basic.BasicDispatcher(*instruments)[source]

Bases: DispatcherProtocol

Dispatches observations to registered instruments.

This class manages: - Finding the appropriate instrument for a given observation. - Dispatching announcements to the relevant instruments.

Parameters:: *instruments (_Instrument) – Variable number of instrument instances.

Example

>>> from abc import ABC, abstractmethod
>>>
>>> class BaseInstrument(ABC):
...     @abstractmethod
...     def add(self):
...         pass
...
>>> class LoggerInstrument:
...     @staticmethod
...     def add():
...         return "Log message added!"
...
>>> class AnalyticsInstrument:
...     @staticmethod
...     def add():
...         return "Analytics entry added!"
...
>>> dispatcher = BasicDispatcher(LoggerInstrument(), AnalyticsInstrument())
>>> dispatcher
BasicDispatcher(instruments=('<domprob.dispatchers.basic.LoggerInstrument object at 0x...>', '<domprob.dispatchers.basic.AnalyticsInstrument object at 0x...>'))
>>>
>>> from domprob import announcement, BaseObservation
>>>
>>> class SomeObservation(BaseObservation):
...     @announcement(LoggerInstrument)
...     @announcement(AnalyticsInstrument)
...     def foo(self, instrument: BaseInstrument) -> None:
...         print(instrument.add())
...
>>> obs = SomeObservation()
>>> dispatcher.dispatch(obs)
Analytics entry added!
Log message added!

__eq__(other)[source]

Return self==value.

Return type:: bool

__repr__()[source]

Return a string representation of the dispatcher.

Returns:

A string representation of the dispatcher and its: instruments.

Return type:

str

Example

>>> from abc import ABC, abstractmethod
>>>
>>> class BaseInstrument(ABC):
...     @abstractmethod
...     def add(self):
...         pass
...
>>> class LoggerInstrument:
...     @staticmethod
...     def add():
...         return "Log message added!"
...
>>> class AnalyticsInstrument:
...     @staticmethod
...     def add():
...         return "Analytics entry added!"
...
>>> dispatcher = BasicDispatcher(LoggerInstrument(), AnalyticsInstrument())
>>> repr(dispatcher)
"BasicDispatcher(instruments=('<domprob.dispatchers.basic.LoggerInstrument object at 0x...>', '<domprob.dispatchers.basic.AnalyticsInstrument object at 0x...>'))"

_abc_impl = <_abc._abc_data object>

_dispatch_ann(observation, announcement)[source]

Process an announcement by identifying the required instrument.

This method retrieves the correct instrument instance based on the announcement and calls _instrum_announce.

Parameters:

observation (_Obs) – The observation being processed.
announcement (_Ann) – The announcement method to invoke.

Return type:

None

static _dispatch_instrum_ann(observation, announcement, instrument)[source]

Invoke an announcement method on an instrument.

This method triggers the specified announcement method on the given instrument instance if it is available.

Parameters:

observation (_Obs) – The observation being processed.
announcement (_Ann) – The announcement method to invoke.
instrument (_Instrument | None) – The target instrument instance, if available.

Return type:

None

_is_protocol = False

_is_runtime_protocol = True

dispatch(observation)[source]

Dispatch an observation to all applicable instruments.

This method retrieves all announcements from the observation and processes them.

Parameters:: observation (_Obs) – The observation to process.
Return type:: None

Example

>>> from abc import ABC, abstractmethod
>>>
>>> class BaseInstrument(ABC):
...     @abstractmethod
...     def add(self):
...         pass
...
>>> class LoggerInstrument:
...     @staticmethod
...     def add():
...         return "Log message added!"
...
>>> class AnalyticsInstrument:
...     @staticmethod
...     def add():
...         return "Analytics entry added!"
...
>>> dispatcher = BasicDispatcher(LoggerInstrument(), AnalyticsInstrument())
>>> dispatcher
BasicDispatcher(instruments=('<domprob.dispatchers.basic.LoggerInstrument object at 0x...>', '<domprob.dispatchers.basic.AnalyticsInstrument object at 0x...>'))
>>>
>>> from domprob import announcement, BaseObservation
>>>
>>> class SomeObservation(BaseObservation):
...     @announcement(LoggerInstrument)
...     @announcement(AnalyticsInstrument)
...     def foo(self, instrument: BaseInstrument) -> None:
...         print(instrument.add())
...
>>> obs = SomeObservation()
>>> dispatcher.dispatch(obs)
Analytics entry added!
Log message added!

class domprob.dispatchers.basic.InstrumentImpRegistry(*instruments)[source]

Bases: Collection[_Instrument]

Registry for instrument implementations, allowing lookup and caching.

This class acts as a collection that stores instruments and supports:

Efficient retrieval of instruments by type.
Caching of previously looked-up instruments for performance optimization.

Parameters:: *instruments (_Instrument) – Variable number of instrument instances to store.

Example

>>> class LoggerInstrument:
...     @staticmethod
...     def add():
...         return "Log message added!"
...
>>> class AnalyticsInstrument:
...     @staticmethod
...     def add():
...         return "Analytics entry added!"
...
>>> logger = LoggerInstrument()
>>> analytics = AnalyticsInstrument()
>>>
>>> registry = InstrumentImpRegistry(logger, analytics)
>>> logger_ = registry.get(LoggerInstrument)
>>>
>>> logger == logger_
True
>>> print(registry.get(object))
None

__contains__(item)[source]

Check if an instrument exists in the registry.

Parameters:: item (object) – The instrument instance or class to check.
Returns:: True if the instrument is present, otherwise False.
Return type:: bool

Example

>>> class LoggerInstrument:
...     @staticmethod
...     def add():
...         return "Log message added!"
...
>>> class AnalyticsInstrument:
...     @staticmethod
...     def add():
...         return "Analytics entry added!"
...
>>> logger = LoggerInstrument()
>>> analytics = AnalyticsInstrument()
>>>
>>> registry = InstrumentImpRegistry(logger, analytics)
>>> logger in registry
True
>>> object in registry
False

__iter__()[source]

Iterate over stored instruments.

Returns:: An iterator over the instruments.
Return type:: Iterator[_Instrument]

Example

>>> class LoggerInstrument:
...     @staticmethod
...     def add():
...         return "Log message added!"
...
>>> class AnalyticsInstrument:
...     @staticmethod
...     def add():
...         return "Analytics entry added!"
...
>>> logger = LoggerInstrument()
>>> analytics = AnalyticsInstrument()
>>>
>>> registry = InstrumentImpRegistry(logger, analytics)
>>>
>>> for instrument in registry:
...     print(instrument.add())
...
Log message added!
Analytics entry added!

__len__()[source]

Return the number of stored instruments.

Returns:: The number of instruments in the registry.
Return type:: int

Example

>>> class LoggerInstrument:
...     @staticmethod
...     def add():
...         return "Log message added!"
...
>>> class AnalyticsInstrument:
...     @staticmethod
...     def add():
...         return "Analytics entry added!"
...
>>> logger = LoggerInstrument()
>>> analytics = AnalyticsInstrument()
>>>
>>> registry = InstrumentImpRegistry(logger, analytics)
>>>
>>> len(registry)
2

__repr__()[source]

Return a string representation of the registry.

Returns:: The string representation of the registry.
Return type:: str

_abc_impl = <_abc._abc_data object>

static _is_hashable(obj)[source]

Check if an object is hashable.

Parameters:: obj (Any) – The object to check.
Returns:: True if the object is hashable, False otherwise.
Return type:: bool

get(instrument_cls, required=False)[source]

Retrieve an instrument instance by its class type.

If the instrument class is hashable, results are cached for efficiency.

Parameters:

instrument_cls (type[TypeVar(_Instrument, bound= Any)]) – The class type of the instrument to retrieve.
required (bool) – If True, raises a KeyError if the instrument is not found. If False, returns None.

Returns:

The retrieved instrument instance or: None if not found.

Return type:

_Instrument | None

Raises:

KeyError – If required is True and the instrument is not found.

Example

>>> class LoggerInstrument:
...     @staticmethod
...     def add():
...         return "Log message added!"
...
>>> class AnalyticsInstrument:
...     @staticmethod
...     def add():
...         return "Analytics entry added!"
...
>>> logger = LoggerInstrument()
>>> analytics = AnalyticsInstrument()
>>>
>>> registry = InstrumentImpRegistry(logger)
>>>
>>> registry.get(LoggerInstrument).add()
'Log message added!'
>>> registry.get(AnalyticsInstrument, required=True)
Traceback (most recent call last):
    ...
KeyError: 'Instrument `AnalyticsInstrument` not found in available implementations: `<domprob.dispatchers.basic.LoggerInstrument object at 0x...>`'

exception domprob.dispatchers.basic.ReqInstrumException(observation, announcement, req_supp_instrum, *instrum_imps)[source]

Bases: DispatcherException

Exception raised when a required instrument is missing an implementation of the same type for an observation announcement.

An instrument is marked as required with the required flag in the @announcement decorator:

>>> from domprob import announcement, BaseObservation
>>>
>>> class SomeObservation(BaseObservation):
...
...     @announcement(..., required=True)
...     def some_method(self, instrument: ...) -> None:
...         ...
...

Parameters:

observation (_Obs) – The observation instance where the missing instrument was required.
announcement (_Ann) – The announcement method that failed due to the missing instrument.
req_supp_instr (type[_Instrument]) – The instrument type that was expected but not found.
*instrum_imps (_Instrument) – The available instrument instances at the time of the failure.

__repr__(): Return repr(self).

add_note(object, /): Exception.add_note(note) – add a note to the exception

args

property msg: str

Constructs a descriptive error message for the exception.

Returns:

A formatted string detailing the missing instrument,: the observation method where it was required, and the available instrument implementations.

Return type:

str

with_traceback(object, /): Exception.with_traceback(tb) – set self.__traceback__ to tb and return self.

exception domprob.dispatchers.dispatcher.DispatcherException[source]

Bases: DomprobException

Base exception for errors occurring within dispatchers.

This exception is raised when an error occurs while processing observations within a dispatcher.

It inherits from DomprobException, allowing it to be caught alongside other domain-specific exceptions.

__repr__(): Return repr(self).

add_note(object, /): Exception.add_note(note) – add a note to the exception

args

with_traceback(object, /): Exception.with_traceback(tb) – set self.__traceback__ to tb and return self.

class domprob.dispatchers.dispatcher.DispatcherProtocol(*args, **kwargs)[source]

Bases: Protocol

Protocol defining the structure for dispatchers handling domain observations.

Classes implementing this protocol must define:

dispatch(): Processes an ObservationProtocol and returns a result.
__repr__(): Provides a string representation of the dispatcher.

This protocol is @runtime_checkable, meaning isinstance(dispatcher, DispatcherProtocol) can be used to verify conformance at runtime.

Example

>>> from domprob.dispatchers.dispatcher import DispatcherProtocol
>>> from domprob.observations.observation import ObservationProtocol
>>>
>>> class ConcreteDispatcher:
...     @staticmethod
...     def dispatch(self, observation: ObservationProtocol) -> str:
...         return "Processed observation"
...
...     def __repr__(self) -> str:
...         return "ConcreteDispatcher()"
...
>>> dispatcher = ConcreteDispatcher()
>>> assert isinstance(dispatcher, DispatcherProtocol)

_abc_impl = <_abc._abc_data object>

_is_protocol = True

_is_runtime_protocol = True

dispatch(observation)[source]

Dispatch an observation and return a result.

Parameters:: observation (ObservationProtocol[_P, _R]) – The observation to process.
Returns:: The result of processing the observation.
Return type:: _R

Example

>>> from domprob import announcement, BaseObservation
>>> from domprob.dispatchers.dispatcher import DispatcherProtocol
>>> from domprob.observations.observation import ObservationProtocol
>>>
>>> class MyDispatcher:
...     @staticmethod
...     def dispatch(observation: ObservationProtocol) -> str:
...         return "Handled"
...
>>> dispatcher = MyDispatcher()
>>>
>>> class SomeInstrument:
...     pass
...
>>> class Obs(BaseObservation):
...     @announcement(SomeInstrument)
...     def foo(self, instrument: SomeInstrument) -> str:
...         pass
...
>>> result = dispatcher.dispatch(Obs())
>>> print(result)
Handled