API Documentation

Event Processor

class src.event_processor.event_processor.EventProcessor(invocation_strategy: src.event_processor.invocation_strategies.InvocationStrategies = <InvocationStrategies.FIRST_MATCH: <class 'src.event_processor.invocation_strategies.FirstMatch'>>)

A self-contained event processor.

add_subprocessor(subprocessor: src.event_processor.event_processor.EventProcessor)

Add a subprocessor to this event processor

Parameters

subprocessor – The other event processor to add

invoke(event: Dict)Any

Invoke the correct processor for an event.

There may be multiple processors invoked, depending on the invocation strategy.

Parameters

event – The event to find a processor for

Returns

The return value of the processor

processor(event_filter: src.event_processor.filters.Filter, rank: int = 0)

Register a new processor with the given filter and rank.

Parameters
  • event_filter – The filter for which to match events

  • rank – This processor’s rank (when there are multiple matches for a single event)

Exceptions

Exceptions for event processor.

exception EventProcessorError

General exception for the event-processor library.

exception FilterError

Exception for failures related to filters.

exception InvocationError

Exception for failures in invocation.

Filtering

Dependency Injection

Dependency injection and management facilities.

class dependencies.Depends(callable_: Callable, cache: bool = True)

Class to designate a dependency

class dependencies.Event(dict_event: dict)

Type to wrap a dict to be used as a dependency.

dependencies.call_with_injection(callable_: Callable, event: Optional[dependencies.Event] = None, cache: Optional[dict] = None, *args, **kwargs)Optional[Any]

Call a callable and inject required dependencies.

Note that keyword args that have the same name as the parameter used for a dependency will be overwritten with the dependency’s injected value.

Parameters
  • callable – The callable to call

  • event – The event for the current invocation

  • cache – The dependency cache to use

  • args – The args to pass to the callable

  • kwargs – The kwargs to pass to the callable

Returns

The return value of the callable

dependencies.get_event_dependencies(callable_: Callable)List[str]

Get the parameter names for event dependencies.

Parameters

callable – The callable for which to get dependencies

Returns

A list of the parameters requiring the event

dependencies.get_required_dependencies(callable_: Callable)Dict[str, dependencies.Depends]

Get the required dependencies for a callable.

Parameters

callable – The callable for which to get dependencies

Returns

A mapping of callable argument names to dependencies

dependencies.resolve(dependency: dependencies.Depends, event: Optional[dependencies.Event] = None, cache: Optional[dict] = None)Tuple[Optional[Any], bool]

Resolve a dependency into a value.

The resulting values from dependencies are cached and re-used if a cache is supplied and the dependency itself does not explicitly state that it does not want to be cached. Also, any dependency that depends on another dependency where caching has been disabled will also not be cached (because the sub-value may change, which may in turn change the value of the current dependency).

Parameters
  • dependency – The dependency to resolve

  • event – The event for the current invocation

  • cache – The cache for previously resolved dependencies

Returns

The tuple (resolved_value, cacheable)