Signals

In Edgy, signals provide a mechanism to "listen" to model events, triggering specific actions when events like saving or deleting occur. This is similar to Django's signals but also draws inspiration from Ormar's implementation, and leverages the blinker library for anonymous signals.

What are Signals?

Signals are used to execute custom logic when certain events happen within Edgy models. They enable decoupling of concerns, allowing you to perform actions like sending notifications, updating related data, or logging events without cluttering your model definitions.

Default Signals

Edgy provides default signals for common model lifecycle events, which you can use out of the box.

How to Use Them

The default signals are located in edgy.core.signals. Import them as follows:

from edgy.core.signals import (
    post_delete,
    post_save,
    post_update,
    pre_delete,
    pre_save,
    pre_update,
)

pre_save

Triggered before a model is saved (during Model.save() and Model.query.create()).

pre_save(send: Type["Model"], instance: "Model")

post_save

Triggered after a model is saved (during Model.save() and Model.query.create()).

post_save(send: Type["Model"], instance: "Model")

pre_update

Triggered before a model is updated (during Model.update() and Model.query.update()).

pre_update(send: Type["Model"], instance: "Model")

post_update

Triggered after a model is updated (during Model.update() and Model.query.update()).

post_update(send: Type["Model"], instance: "Model")

pre_delete

Triggered before a model is deleted (during Model.delete() and Model.query.delete()).

pre_delete(send: Type["Model"], instance: "Model")

post_delete

Triggered after a model is deleted (during Model.delete() and Model.query.delete()).

post_update(send: Type["Model"], instance: "Model")

Receiver

A receiver is a function that executes when a signal is triggered. It "listens" for a specific event.

Example: Given the following model:

import edgy

database = edgy.Database("sqlite:///db.sqlite")
registry = edgy.Registry(database=database)


class User(edgy.Model):
    id: int = edgy.BigIntegerField(primary_key=True)
    name: str = edgy.CharField(max_length=255)
    email: str = edgy.CharField(max_length=255)
    is_verified: bool = edgy.BooleanField(default=False)

    class Meta:
        registry = registry

You can send an email to a user upon creation using the post_save signal:

from edgy.core.signals import post_save


def send_notification(email: str) -> None:
    """
    Sends a notification to the user
    """
    send_email_confirmation(email)


@post_save.connect_via(User)
async def after_creation(sender, instance, **kwargs):
    """
    Sends a notification to the user
    """
    send_notification(instance.email)

The @post_save decorator specifies the User model, indicating it listens for events on that model.

Requirements

Receivers must meet the following criteria:

• Must be a callable (function).
• Must have sender as the first argument (the model class).
• Must have **kwargs to accommodate changes in model attributes.
• Must be async to match Edgy's async operations.

Multiple Receivers

You can use the same receiver for multiple models:

import edgy

database = edgy.Database("sqlite:///db.sqlite")
registry = edgy.Registry(database=database)


class User(edgy.Model):
    id: int = edgy.BigIntegerField(primary_key=True)
    name: str = edgy.CharField(max_length=255)
    email: str = edgy.CharField(max_length=255)

    class Meta:
        registry = registry


class Profile(edgy.Model):
    id: int = edgy.BigIntegerField(primary_key=True)
    profile_type: str = edgy.CharField(max_length=255)

    class Meta:
        registry = registry

from edgy.core.signals import post_save


def send_notification(email: str) -> None:
    """
    Sends a notification to the user
    """
    send_email_confirmation(email)


@post_save.connect_via(User)
@post_save.connect_via(Profile)
async def after_creation(sender, instance, **kwargs):
    """
    Sends a notification to the user
    """
    if isinstance(instance, User):
        send_notification(instance.email)
    else:
        # something else for Profile
        ...

Multiple Receivers for the Same Model

You can have multiple receivers for the same model:

from edgy.core.signals import post_save


def push_notification(email: str) -> None:
    # Sends a push notification
    ...


def send_email(email: str) -> None:
    # Sends an email
    ...


@post_save.connect_via(User)
async def after_creation(sender, instance, **kwargs):
    """
    Sends a notification to the user
    """
    send_email(instance.email)


@post_save.connect_via(User)
async def do_something_else(sender, instance, **kwargs):
    """
    Sends a notification to the user
    """
    push_notification(instance.email)

Disconnecting Receivers

You can disconnect a receiver to prevent it from running:

from edgy.core.signals import post_save


def send_notification(email: str) -> None:
    """
    Sends a notification to the user
    """
    send_email_confirmation(email)


@post_save.connect_via(User)
async def after_creation(sender, instance, **kwargs):
    """
    Sends a notification to the user
    """
    send_notification(instance.email)


# Disconnect the given function
User.meta.signals.post_save.disconnect(after_creation)

Custom Signals

Edgy allows you to define custom signals, extending beyond the default ones.

Continuing with the User model example:

import edgy

database = edgy.Database("sqlite:///db.sqlite")
registry = edgy.Registry(database=database)


class User(edgy.Model):
    id: int = edgy.BigIntegerField(primary_key=True)
    name: str = edgy.CharField(max_length=255)
    email: str = edgy.CharField(max_length=255)
    is_verified: bool = edgy.BooleanField(default=False)

    class Meta:
        registry = registry

Create a custom signal named on_verify:

import edgy
from edgy.core import signals
from edgy import Signal
# or:
# from blinker import Signal

database = edgy.Database("sqlite:///db.sqlite")
registry = edgy.Registry(database=database)


class User(edgy.Model):
    id: int = edgy.BigIntegerField(primary_key=True)
    name: str = edgy.CharField(max_length=255)
    email: str = edgy.CharField(max_length=255)

    class Meta:
        registry = registry


# Create the custom signal
User.meta.signals.on_verify = Signal()

The on_verify signal is now available for the User model.

Danger

Signals are class-level attributes, affecting all derived instances. Use caution when creating custom signals.

Create a receiver for the custom signal:

import edgy

from edgy import Signal
# or:
# from blinker import Signal

database = edgy.Database("sqlite:///db.sqlite")
registry = edgy.Registry(database=database)


class User(edgy.Model):
    id: int = edgy.BigIntegerField(primary_key=True)
    name: str = edgy.CharField(max_length=255)
    email: str = edgy.CharField(max_length=255)

    class Meta:
        registry = registry


# Create the custom signal
User.meta.signals.on_verify = Signal()


# Create the receiver
async def trigger_notifications(sender, instance, **kwargs):
    """
    Sends email and push notification
    """
    send_email(instance.email)
    send_push_notification(instance.email)


# Register the receiver into the new Signal.
User.meta.signals.on_verify.connect(trigger_notifications)

The trigger_notifications receiver is now connected to the on_verify signal.

Rewire Signals

To prevent default lifecycle signals from being called, you can overwrite them per class or use the set_lifecycle_signals_from method of the Broadcaster:

import edgy

from edgy import Signal
# or:
# from blinker import Signal

database = edgy.Database("sqlite:///db.sqlite")
registry = edgy.Registry(database=database)


class User(edgy.Model):
    id: int = edgy.BigIntegerField(primary_key=True)
    name: str = edgy.CharField(max_length=255)
    email: str = edgy.CharField(max_length=255)

    class Meta:
        registry = registry


# Overwrite a model lifecycle Signal; this way the main signals.pre_delete is not triggered
User.meta.signals.pre_delete = Signal()

# Update all lifecyle signals. Replace pre_delete again with the default
User.meta.signals.set_lifecycle_signals_from(signals)

How to Use It

Use the custom signal in your logic:

async def create_user(**kwargs):
    """
    Creates a user
    """
    await User.query.create(**kwargs)


async def is_verified_user(id: int):
    """
    Checks if user is verified and sends notification
    if true.
    """
    user = await User.query.get(pk=id)

    if user.is_verified:
        # triggers the custom signal
        await User.meta.signals.on_verify.send_async(User, instance=user)

The on_verify signal is triggered only when the user is verified.

Disconnect the Signal

Disconnecting a custom signal is the same as disconnecting a default signal:

async def trigger_notifications(sender, instance, **kwargs):
    """
    Sends email and push notification
    """
    send_email(instance.email)
    send_push_notification(instance.email)


# Disconnect the given function
User.meta.signals.on_verify.disconnect(trigger_notifications)