Events

Introduction

Almasix’s application event bus lives in almasix.events. It is separate from Articulate model events (creating, saved, …). Use it to decouple domain actions from side effects.

from almasix.events import Event, event, listen

class OrderShipped:
    def __init__(self, order_id: int) -> None:
        self.order_id = order_id

def send_receipt(event: OrderShipped) -> None:
    ...

Event.listen(OrderShipped, send_receipt)
Event.dispatch(OrderShipped(42))
# or: event(OrderShipped(42))

Registering listeners

Register in a provider boot() method, or subclass EventServiceProvider and fill the listen map:

from almasix.events import Event, EventServiceProvider

class AppEventServiceProvider(EventServiceProvider):
    listen = {
        OrderShipped: [SendShipmentNotification],
    }

Closure listeners and class listeners (with a handle method) are both supported. String class paths resolve via import.

Wildcards

Event.listen("orders.*", lambda name, payload: ...)

Wildcard listeners receive (event_name, payload).

Subscribers

class OrderSubscriber:
    def subscribe(self, events) -> None:
        events.listen(OrderShipped, self.on_shipped)

    def on_shipped(self, event: OrderShipped) -> None:
        ...

Event.subscribe(OrderSubscriber)

Dispatching

Event.dispatch(OrderShipped(1))
Event.until("ping")          # halt on first non-None response
event(OrderShipped(1))       # helper

Return False from a listener to stop propagation.

Queued listeners

Implement ShouldQueue (from almasix.events / almasix.queue) on a listener class. When the event fires, Almasix pushes a CallQueuedListener job:

from almasix.events import ShouldQueue

class SendShipmentNotification(ShouldQueue):
    queue = "listeners"
    delay = 0

    def handle(self, event: OrderShipped) -> None:
        ...

Optional should_queue(event) -> bool, via_connection(), via_queue(), and with_delay(event) mirror the Laravel surface.

Generating stubs

smith make:event OrderShipped
smith make:listener SendShipmentNotification --event=OrderShipped --queued
smith event:list

Broadcasting

An event that inherits ShouldBroadcast also leaves the server: dispatch puts it on the channels it names, before your listeners run, and the browser hears about it over a websocket. See Broadcasting.

from almasix.broadcasting import PrivateChannel, ShouldBroadcast


class OrderShipped(ShouldBroadcast):
    def broadcast_on(self):
        return [PrivateChannel(f"orders.{self.order_id}")]

Testing

Event.fake()
Event.dispatch(OrderShipped(1))
Event.assert_dispatched(OrderShipped)
Event.assert_not_dispatched(OtherEvent)
Event.assert_nothing_dispatched()  # after a fresh fake()