Events

PERSEUS includes a lightweight event system built on top of blinker. It allows custom states and services to react to changes in the system — such as a project being created or updated — without tight coupling between components.

DatabaseItem events

All classes that extend DatabaseItem (e.g. Project) automatically emit signals when they are created, updated, or deleted through the database manager. These signals follow a predictable naming convention:

Event	Signal name
Created	`ClassName.add`
Updated	`ClassName.update`
Deleted	`ClassName.delete`

For Project, PERSEUS additionally emits attribute-specific update signals whenever a single field changes:

Event	Signal name
A specific field changed	`Project.update.<field_name>`

The payload for attribute-specific signals contains:

object — the updated Project instance
attribute — name of the changed field
old — the previous field value
new — the new field value

Listening to events in a service

Inside a BaseService subclass, use the @BaseService.on_event(signal_name) decorator to register a class method as an event handler. The handler is registered automatically when PERSEUS loads the service.

The first argument after cls receives the object from the signal payload (the DatabaseItem that triggered the event). All remaining payload entries are passed as keyword arguments.

from typing import Any

from perseus.datamanager import Project

from .. import BaseService


class NotificationService(BaseService):
    @classmethod
    def initialize(cls):
        pass

    @BaseService.on_event("Project.add")
    @classmethod
    def on_project_created(cls, project: Project):
        # Called whenever a Project is saved to the database for the first time
        print(f"New project created: {project.abbreviation}")

    @BaseService.on_event("Project.update")
    @classmethod
    def on_project_updated(cls, project: Project, old: Project):
        # Called whenever any field of a Project changes
        print(f"Project updated: {project.abbreviation}")

    @BaseService.on_event("Project.update.state_machine")
    @classmethod
    def on_project_statemachine_changed(
        cls, project: Project, attribute: str, old: Any, new: Any
    ):
        # Called only when the `state_machine` attribute of a Project changes
        print(
            f"Project {project.abbreviation} change the state_machine from {old!r} to {new!r}"
        )

    @BaseService.on_event("Project.delete")
    @classmethod
    def on_project_deleted(cls, project: Project):
        print(f"Project deleted: {project.abbreviation}")

Listening to events outside a service

The @receiver decorator from perseus.utils.signals connects any callable to a named signal. This is useful in custom states or standalone helper modules.

from perseus.datamanager import Project
from perseus.utils.signals import receiver


@receiver("Project.add")
def on_project_created(sender, payload=None):
    project: Project = payload["object"] if payload else sender
    print(f"New project created: {project.abbreviation}")

Custom signals

You can define your own signals to communicate between services or states. Use create_and_send_signal to emit a signal and @receiver (or @BaseService.on_event) to handle it.

Emitting a custom signal:

from perseus.utils.signals import create_and_send_signal

MY_SIGNAL = "MyService.job_finished"


def finish_job(result: dict):
    # ... do work ...
    create_and_send_signal(
        sender=MyService,
        signal_name=MY_SIGNAL,
        payload={"result": result},
    )

Receiving a custom signal:

from perseus.utils.signals import receiver

MY_SIGNAL = "MyService.job_finished"


@receiver(MY_SIGNAL)
def on_job_finished(sender, payload=None):
    result = payload.get("result") if payload else None
    print(f"Job finished with result: {result}")