Skip to content

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 (
    pre_delete,
    post_delete,
    pre_save,
    post_save,
    pre_update,
    post_update,
    post_migrate,
    pre_migrate,
    pre_relation_add,
    post_relation_add,
    pre_relation_remove,
    post_relation_remove,
    pre_bulk,
    post_bulk,
)

Pre Operations special exception

If you just want to skip the operation without causing a bigger error, you can raise edgy.exceptions.SkipOperation to stop the operation and returning an empty value and send the corresponding post signal with operation_skipped=True.

pre_save

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

pre_save(send: type["Model"], instance: Union["Model", "QuerySet"], 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: Union["Model", "QuerySet"], 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"], model_instance: Optional["Model"], 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"], model_instance: Optional["Model"], 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.
  • model_instance -The model instance if available. For save signals always available
  • 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.
  • row_count - (post only) The rows updated.
  • operation_skipped - (post only) If the operation was skipped.

pre_delete

Triggered before a model is deleted (during Model.delete() and Model.query.delete()). Exception: intermediate model instances when using relations.

pre_delete(send: type["Model"], instance: Union["Model", "QuerySet"], model_instance: Optional["Model"])

A more advanced example is:

import edgy


class BaseModel(edgy.StrictModel):
    protected = edgy.BooleanField(default=False)
    __deletion_with_signals__ = True
    __require_model_based_deletion__ = True

    class Meta:
        registry = ...
        abstract = True


class Friend(BaseModel):
    name = edgy.CharField(max_length=100)


class Profile(BaseModel):
    name = edgy.CharField(max_length=100)


class User(BaseModel):
    name = edgy.CharField(max_length=100)
    profile = edgy.ForeignKey("Profile", null=True, on_delete=edgy.CASCADE, related_name="users")
    friends = edgy.ManyToMany("Friend", related_name="users")


@User.meta.signals.pre_relation_remove.connect_via(User)
@Friend.meta.signals.pre_relation_remove.connect_via(Friend.meta.fields["users"].through)
@Profile.meta.signals.pre_relation_remove.connect_via(Profile)
async def abort_removal_relation(sender, raw_values, **kwargs):
    for value in raw_values:
        if value.protected:
            raise Exception()


@User.meta.signals.pre_delete.connect_via(User)
@Friend.meta.signals.pre_delete.connect_via(Friend.meta.fields["users"].through)
@Profile.meta.signals.pre_delete.connect_via(Profile)
async def abort_removal_relation(sender, model_instance, **kwargs):
    if model_instance.protected:
        raise Exception()
pre_delete parameters
  • instance - The model or QuerySet instance.
  • model_instance - The model instance if available otherwise None.
  • injected_filters (query only) - You can insert or remove (when inserted by another signal) extra filter parameters for query deletions.

Example for the insertion of new parameters

import edgy
from edgy.exceptions import SkipOperation


class BaseModel(edgy.StrictModel):
    protected = edgy.BooleanField(default=False)

    class Meta:
        registry = ...
        abstract = True


class Friend(BaseModel):
    name = edgy.CharField(max_length=100)


class Profile(BaseModel):
    name = edgy.CharField(max_length=100)


class User(BaseModel):
    name = edgy.CharField(max_length=100)
    profile = edgy.ForeignKey("Profile", null=True, on_delete=edgy.CASCADE, related_name="users")
    friends = edgy.ManyToMany("Friend", related_name="users")


@User.meta.signals.pre_relation_remove.connect_via(User)
@Profile.meta.signals.pre_relation_remove.connect_via(Profile)
async def excempt_removal_relation(sender, raw_values, **kwargs):
    new_raw_values = list(raw_values)
    raw_values.clear()
    for value in new_raw_values:
        if not value.protected:
            raw_values.append(value)


@User.meta.signals.pre_delete.connect_via(User)
@Profile.meta.signals.pre_delete.connect_via(Profile)
async def abort_removal_relation(sender, model_instance, injected_filters, **kwargs):
    if model_instance is None:
        injected_filters.append({"protected": True})
    else:
        if model_instance.protected:
            raise SkipOperation()

You can also add or_, and_ or other clauses valid for the filter method of QuerySet. By default they are combined like with and_.

post_delete

Triggered after a model is deleted (during Model.delete() and Model.query.delete()). Exception: through model instances when using relation methods.

post_delete(send: type["Model"], instance: Union["Model", "QuerySet"], model_instance: Optional["Model"], 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).
  • model_instance -The model instance if available.

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.

pre_bulk

The pre_bulk signal is issued before the database modifications and allows before executing bulk operations to manipulate the instances. Sender is the queryset. When using bulk operations on the relation queryset the sender is either the target model (one-to-many) or through model (many-to-many).

import edgy


class BaseModel(edgy.StrictModel):
    active = edgy.IntegerField(default=0)
    duplicate: bool = False

    class Meta:
        registry = ...
        abstract = True


class Friend(BaseModel):
    name = edgy.CharField(max_length=100)


class Profile(BaseModel):
    name = edgy.CharField(max_length=100)


class User(BaseModel):
    name = edgy.CharField(max_length=100)
    profile = edgy.ForeignKey("Profile", null=True, on_delete=edgy.CASCADE, related_name="users")
    friends = edgy.ManyToMany("Friend", related_name="users")


@User.meta.signals.pre_bulk.connect_via(User)
@Profile.meta.signals.pre_bulk.connect_via(Profile)
async def handle_active_duplicate_bulk(sender, raw_values, create_params, update_params, **kwargs):
    for item in create_params:
        if item[0].active > 0:
            item[0].active = 0

    for item in update_params:
        if item[0].active > 0:
            item[0].active -= 1
        if item[0].duplicate:
            create_params.append((item[0].model_copy(), item[1], item[2]))

post_bulk

The post_bulk signal is issued after the database modifications and contains information about how many rows were created and/or updated. When using bulk operations on the relation queryset the sender is either the target model (one-to-many) or through model (many-to-many).

Parameters of *_bulk signals

  • raw_values: Raw model instances with created flag. No resolving of embed_parent.
  • values (post only, only when operation_skipped=False): Resolved model instances with created flag. When not using resolve_embed, the raw model instances.
  • operation: bulk_create, bulk_update, bulk_update_or_create, bulk_get_or_create.
  • resolve_embed: Value of resolve_embed.
  • create_params: (raw_instance, position in raw_values, set of input kwarg names) tuple. You can prevent an insert operation by removing an tuple. You can move an tuple to update_params if the instance should be updated instead.
  • update_params: (raw_instance, position in raw_values, set of input kwarg names) tuple. You can prevent an update operation by removing an tuple. You can move an tuple to create_params if the instance should be created instead (not recommended!).
  • ignore_conflicts (for bulk_create): Value of ignore_conflicts.
  • update_fields (for updating bulk operations): The fields updated.
  • row_count_update (for updating bulk operations): Amount of rows updated. None for db systems not supporting it.
  • row_count_create (for creating bulk operations): Amount of rows created. None for db systems not supporting it.

create_params and update_params

Theoretical you can change the logic of what is inserted or updated but this is not recommended especially because you are restricted to the model type and by the update_fields which you are not able to change. You can however change the identifying_db_fields and move an instance to the update list or just remove it completely from both lists so it is only shown. Or for methods allowing None in instance slots, even replace the raw_values slot with None. Be careful with the last trick! You can break core assumptions, so we can have None instances where typings forbid this. Methods where this trick can be applied are: bulk_create (with ignore_conflicts=True), bulk_update, bulk_update_or_create.

pre_relation_add

The pre_relation_add signal is issued before the database modifications and allows before executing changing the relations to manipulate the instances.

The sender is either the target model (one-to-many) or through model (many-to-many). Signals are also issued on overwrites of pre_relation_add in source.meta.signals, through.meta.signals and target.meta.signals. However if the default signal or a shared signal is used it is only issued per different signal object, so you can expect when listening to one of the signals, you get notified only once.

Note

This signal is not issued if add_many is called with empty arguments or save_related is called when staged are empty.

post_relation_add

The post_relation_add signal is issued after the database modifications and contains information about how many rows were changed.

The sender is either the target model (one-to-many) or through model (many-to-many). Signals are also issued on overwrites of pre_relation_add in source.meta.signals, through.meta.signals and target.meta.signals. However if the default signal or a shared signal is used it is only issued per different signal object, so you can expect when listening to one of the signals, you get notified only once.

Note

This signal is not issued if add_many is called with empty arguments or save_related is called when staged are empty.

Parameters of *_relation_add signals

  • instance: Source instance.
  • row_count: How many rows were updated/created? None for db systems not supporting it.
  • row_count_create: How many rows were created? None for db systems not supporting it.
  • raw_values: Raw model instances with created flag of either the source model (one_to_many) or the through model (many_to_many). Useful in combination with create_params and update_params to tweak output. There is no resolving via resolve_embed
  • values (post only, only when add, add_many and operation_skipped=False): The resolved counterpart instances with created flag.
  • operation: save_related and add (also issued for add_many and create).
  • field: RelationField name on source triggering this signal.
  • source: Source model which contains the RelationField triggering the signals.
  • target: Target model.
  • relation: Relation type. one_to_many, many_to_many.
  • create_params: See bulk signal parameter.
  • update_params: See bulk signal parameter.
  • operation_skipped: Is the operation skipped? This will lead to missing parameters (values)

Replacing raw_values/values

Here every operation allows None values instead of instances. So it is no problem to replace in raw_values (pre) or values (post) with None signalling a missing instance/failed update.

pre_relation_remove

The pre_relation_remove signal is issued before the database modifications for the removal of connections and allow customizations including to stop the deletion by issuing an exception.

The sender is either the target model (one-to-many) or through model (many-to-many). Signals are also issued on overwrites of pre_relation_add in source.meta.signals, through.meta.signals and target.meta.signals. However if the default signal or a shared signal is used it is only issued per different signal object, so you can expect when listening to one of the signals, you get notified only once.

Note

This signal is not issued if remove_many is called with empty arguments.

Remove from removal list

You have two options to block the removal of an instance 1. raise an exception 2. remove the instance from raw_values

import edgy


class BaseModel(edgy.StrictModel):
    protected = edgy.BooleanField(default=False)
    __deletion_with_signals__ = True
    __require_model_based_deletion__ = True

    class Meta:
        registry = ...
        abstract = True


class Friend(BaseModel):
    name = edgy.CharField(max_length=100)


class Profile(BaseModel):
    name = edgy.CharField(max_length=100)


class User(BaseModel):
    name = edgy.CharField(max_length=100)
    profile = edgy.ForeignKey("Profile", null=True, on_delete=edgy.CASCADE, related_name="users")
    friends = edgy.ManyToMany("Friend", related_name="users")


@User.meta.signals.pre_relation_remove.connect_via(User)
@Friend.meta.signals.pre_relation_remove.connect_via(Friend.meta.fields["users"].through)
@Profile.meta.signals.pre_relation_remove.connect_via(Profile)
async def abort_removal_relation(sender, raw_values, **kwargs):
    for value in raw_values:
        if value.protected:
            raise Exception()


@User.meta.signals.pre_delete.connect_via(User)
@Friend.meta.signals.pre_delete.connect_via(Friend.meta.fields["users"].through)
@Profile.meta.signals.pre_delete.connect_via(Profile)
async def abort_removal_relation(sender, model_instance, **kwargs):
    if model_instance.protected:
        raise Exception()

import edgy
from edgy.exceptions import SkipOperation


class BaseModel(edgy.StrictModel):
    protected = edgy.BooleanField(default=False)

    class Meta:
        registry = ...
        abstract = True


class Friend(BaseModel):
    name = edgy.CharField(max_length=100)


class Profile(BaseModel):
    name = edgy.CharField(max_length=100)


class User(BaseModel):
    name = edgy.CharField(max_length=100)
    profile = edgy.ForeignKey("Profile", null=True, on_delete=edgy.CASCADE, related_name="users")
    friends = edgy.ManyToMany("Friend", related_name="users")


@User.meta.signals.pre_relation_remove.connect_via(User)
@Profile.meta.signals.pre_relation_remove.connect_via(Profile)
async def excempt_removal_relation(sender, raw_values, **kwargs):
    new_raw_values = list(raw_values)
    raw_values.clear()
    for value in new_raw_values:
        if not value.protected:
            raw_values.append(value)


@User.meta.signals.pre_delete.connect_via(User)
@Profile.meta.signals.pre_delete.connect_via(Profile)
async def abort_removal_relation(sender, model_instance, injected_filters, **kwargs):
    if model_instance is None:
        injected_filters.append({"protected": True})
    else:
        if model_instance.protected:
            raise SkipOperation()

post_relation_remove

The post_relation_remove signal is issued after the database modifications for the removal of connections and contains information about how many rows were changed/removed.

The sender is either the target model (one-to-many) or through model (many-to-many). Signals are also issued on overwrites of pre_relation_add in source.meta.signals, through.meta.signals and target.meta.signals. However if the default signal or a shared signal is used it is only issued per different signal object, so you can expect when listening to one of the signals, you get notified only once.

Note

This signal is not issued if remove_many is called with empty arguments.

Parameters of *_relation_remove signals

  • instance: Source instance.
  • row_count (post): How many rows were updated/deleted? None for db systems not supporting it.
  • raw_values: Raw model instances without created flag of either the source model (one_to_many) or the through model (many_to_many). There is no resolving via resolve_embed. You can clear the list and readd the models you want to delete (or doing position based modifications (harder)) as long you don't await during the modifications. You should recheck if something changes if you fetch something with await.
  • field: RelationField name on source triggering this signal.
  • source: Source model which contains the RelationField triggering the signals.
  • target: Target model.
  • relation: Relation type. one_to_many, many_to_many.
  • model_based_deletion (many_to_many): Was a model_based_deletion used?

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

Using a custom signal 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)
        # or when maybe a proxy
        await User.meta.signals.on_verify.send_async(User.get_real_class(), instance=user)
The on_verify signal is triggered only when the user is verified.

Log changes

An other useful usecase is logging user actions:

from sqlalchemy import ForeignKey
import edgy
from contextvars import ContextVar
from edgy.core import signals

models = edgy.Registry(...)
current_user = ContextVar("current_user", default=None)


class BaseModel(edgy.StrictModel):
    class Meta:
        registry = models
        abstract = True


class Friend(BaseModel):
    name = edgy.CharField(max_length=100)


class Profile(BaseModel):
    name = edgy.CharField(max_length=100)


class User(BaseModel):
    name = edgy.CharField(max_length=100)
    profile = edgy.ForeignKey("Profile", null=True, on_delete=edgy.CASCADE, related_name="users")
    friends = edgy.ManyToMany("Friend", related_name="users")


class Log(BaseModel):
    signal = edgy.CharField(max_length=255)
    class_name = edgy.CharField(max_length=255)
    params = edgy.JSONField()
    user = edgy.ForeignKey(User, null=True, on_delete=edgy.CASCADE, related_name="logs")

    def __str__(self) -> str:
        return str(self.extract_db_fields())

    def __repr__(self) -> str:
        return f"Log<{self}>"


for signal_name in dir(signals):
    if not signal_name.startswith("post_") or signal_name == "post_migrate":
        continue
    signal: signals.Signal = getattr(signals, signal_name)

    async def log(sender, _signal_name=signal_name, **kwargs):
        await Log.query.create(
            signal=_signal_name,
            class_name=sender.__name__,
            params={k: str(v) for k, v in kwargs.items()},
            user=current_user.get(),
        )

    for model in models.models.values():
        if model is not Log:
            # weak must be False otherwise the receivers vanish
            signal.connect(log, model, weak=False)

Of course there are better ways for serialization.

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)