delegate-events 0.0.0

Python implementation of the Event Pattern

delegate-events: Python implementation of the Event Pattern.

delegate-events provides a basic implementation of the well-known Event or Pub/Sub Pattern, and is built on top of the delegate-pattern package which handles the delegation part.

Example

from delegate.pattern import delegate
from delegate.events import Channel, Event

class OnSomethingEvent(Event):
    def __init__(self, what: str, why: str):
        self.what = what
        self.why = why

    def __str__(self):
        return f"{self.what} happened because of {self.why}"

class Class1:
    on_something = delegate(Channel[OnSomethingEvent])

inst = Class1()
msg_queue: list[Event] = []

def fn(publisher, event: OnSomethingEvent):
    msg_queue.append(event)

inst.on_something.subscribe(fn)
inst.on_something.fire(OnSomethingEvent("Something", "some other thing"))

str(msg_queue[0]) # => "Something happened because of some other thing"

API

Event class

The Event class is the base class of all events. Events are the classes holding the event information, when an event is fired, thus, when subclassed, should contain properties/fields of the event information.

Example

from delegate.events import Event

class OnSomething(Event):
    def __init__(self, what: str, why: str):
        self.what = what
        self.why = why

    def __str__(self):
        return f"{self.what} happened because of {self.why}"

Channel class

The Channel class is the event channel returned by the delegate, and its jobs are handling subscriptions and firing of events.

Channel.subscribe(handler: Callable[[object, T<Event>], None])

The subscribe function subscribes a handler function to the channel. The function takes 1 argument:

• handler: Callable[[object, Event], None]

Example

from delegate.events import Channel

channel = Channel[OnSomethingEvent]()

def handler(sender: object, args: OnSomethingEvent) -> None:
    pass

channel.subscribe(handler)

Channel.unsubscribe(handler: Callable[[object, T<Event>], None])

The unsubscribe function unsubscribes a handler function from the channel. The function takes 1 argument:

• handler: Callable[[object, Event], None]

from delegate.events import Channel

channel = Channel[OnSomethingEvent]()

def handler(sender: object, args: OnSomethingEvent) -> None:
    pass

channel.subscribe(handler)
channel.unsubscribe(handler)

Channel.fire(event: T<Event>, *, ttl: float | None = 0) -> bool function

The fire function fires the event, invoking all subscribed handlers. The function returns True if event was successfully sent and takes 1 argument:

• event: Event

and 1 optional keyword argument:

• ttl: float | None the time-to-live (seconds) of the event.

from delegate.events import Channel

channel = Channel[OnSomethingEvent]()

msg_queue: list[Event] = []

def handler(sender: object, event: OnSomethingEvent) -> None:
    msg_queue.append(event)

channel.fire(OnSomethingEvent("Something", "some other thing")) # => False
channel.fire(OnSomethingEvent("Something", "some other thing"), ttl=10) # => False - event stays in queue and will be sent if subcrribed to within 10 seconds though

channel.subscribe(handler)

channel.fire(OnSomethingEvent("Something", "some other thing")) # => True

len(msg_queue) # => 2