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,
    post_migrate,
    pre_delete,
    pre_save,
    pre_update,
    pre_migrate,
)

pre_save

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

pre_save(send: Type["Model"], instance: "Model", values: dict, column_values: dict, is_update: bool)

Only the EXPLICIT_SPECIFIED_VALUES contextvar is available.

post_save

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

post_save(send: Type["Model"], instance: "Model", values: dict, column_values: dict, is_update: bool)

pre_update

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

pre_update(sender: Type["Model"], instance: Union["Model", "QuerySet"], values: dict, column_values: dict)

post_update

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

post_update(sender: Type["Model"], instance: Union["Model", "QuerySet"], values: dict, column_values: dict)

pre_save, post_save, pre_update, post_update parameters

The receiver function receives following parameters:

• instance - The model or QuerySet instance. For save signals only the model instance is possible.
• values - The passed values.
• column_values - The parsed values which are used for the db.
• is_update - Is it an update? This is also set for *_update to match the save parameters.
• is_migration - Called from apply_default_force_nullable_fields which is mostly for migrations. Here we have model instances.

pre_delete

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

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

pre_delete parameters

• instance - The model or QuerySet instance.

post_delete

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

post_delete(send: Type["Model"], instance: Union["Model", "QuerySet"], row_count: Optional[int])

post_delete parameters

• instance - The model or QuerySet instance.
• row_count - How many rows are deleted (only some dbs can be None).

pre_migrate

Triggered before upgrading, downgrading or creating a migration. This signal is sync but can be used with async receivers too.

It has following senders:

• "upgrade"
• "downgrade"
• "revision"

post_migrate

Triggered after upgrading, downgrading or creating a migration. This signal is sync but can be used with async receivers too.

It has following senders:

• "upgrade"
• "downgrade"
• "revision"

pre_migrate & post_migrate parameters

Basically all parameters which are passed to the alembic function

That are for upgrade/downgrade:

• config - The configuration object.
• revision - The revision to use (relative or absolute).
• sql - Is offline mode (outputs sql).
• tag -Parameters for env.py script.

And for revision:

• config - The configuration object.
• message - The message.
• autogenerate - Shall the migration file be autogenerated?
• sql - Offline mode.
• head - Head parameter of alembic.
• splice - Splice parameter of alembic.
• branch_label
• version_path
• revision_id - Revision id of the migration.

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)