Banshee

Testing

See the requests a bus dispatches.

Traceable bus

A decorator for a Bus that records all requests that it handles. This is useful in testing and debugging a usage of the bus.

Usage

Wrap an existing bus in the traceable bus, and use it as normal.

bus = banshee.TraceableBus(
   banshee.Builder()
   .with_registry()
   .build()
)

await bus.handle(GreetCommand(name="joe"))

print(bus.messages)

Reference

class banshee.TraceableBus(inner)

Bases: Bus

Traceable bus.

A Bus decorator that stores information about the processing of each message passed to the wrapped instance.

Parameters:

inner (Bus) – decorated instance

async handle(request, contexts=None)

Handle.

Process a message.

Parameters:

  • request (T | Message[T]) – request instance

  • contexts (Iterable[Any] | None) – additional context objects

Return type:

Message[T]

property messages: Sequence[MessageInfo]

Messages.

Returns:

sequence containing information about each message processed.

reset()

Reset.

Clear everything in messages.

Return type:

None

class banshee.MessageInfo(request, contexts, filename, function, lineno, timestamp, result_contexts=None)

Bases: object

Message information.

Contains information about the handling of as specific message.

Parameters:

  • request (object) – request object

  • contexts (tuple[object, ...]) – request message contexts

  • filename (str) – filename where call originated

  • function (str) – function where call originated

  • lineno (int) – line in file where call originated

  • timestamp (datetime) – timestamp of call

  • result_contexts (tuple[object, ...] | None) – result message contexts

contexts: tuple[object, ...]

request message contexts

filename: str

filename where call originated

function: str

function where call originated

lineno: int

line in file where call originated

request: object

request object

result_contexts: tuple[object, ...] | None = None

result message contexts

timestamp: datetime

timestamp of call