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)