signal

A signal is a mechanism that triggers your code (function/method) whenever an event of a given type occurs in ormar.

To do this, you need to register the receiver for the given type of signal of the selected model.

Define receiver

The following example model is given:

import databases
import sqlalchemy

import ormar


base_ormar_config = ormar.OrmarConfig(
    database=databases.Database("sqlite:///db.sqlite"),
    metadata=sqlalchemy.MetaData(),
)


class Album(ormar.Model):
    ormar_config = base_ormar_config.copy()

    id: int = ormar.Integer(primary_key=True)
    name: str = ormar.String(max_length=100)
    is_best_seller: bool = ormar.Boolean(default=False)
    play_count: int = ormar.Integer(default=0)

For example, you can define a trigger that sets the album.is_best_seller state if the album has been played more than 50 times.

Import the\pre_update decorator and see the list of currently available decorators/signals below.

import asyncio

import databases
import ormar
import sqlalchemy
from examples import create_drop_database
from ormar import pre_update

DATABASE_URL = "sqlite:///test.db"

ormar_base_config = ormar.OrmarConfig(
    database=databases.Database(DATABASE_URL),
    metadata=sqlalchemy.MetaData(),
)


class Album(ormar.Model):
    ormar_config = ormar_base_config.copy(
        tablename="albums",
    )

    id: int = ormar.Integer(primary_key=True)
    name: str = ormar.String(max_length=100)
    is_best_seller: bool = ormar.Boolean(default=False)
    play_count: int = ormar.Integer(default=0)


@pre_update(Album)
async def before_update(sender, instance, **kwargs):
    if instance.play_count > 50 and not instance.is_best_seller:
        instance.is_best_seller = True


@create_drop_database(base_config=ormar_base_config)
async def run_query():
    # here album.play_count ans is_best_seller get default values
    album = await Album.objects.create(name="Venice")
    assert not album.is_best_seller
    assert album.play_count == 0

    album.play_count = 30
    # here a trigger is called but play_count is too low
    await album.update()
    assert not album.is_best_seller

    album.play_count = 60
    await album.update()
    assert album.is_best_seller


asyncio.run(run_query())

Define your function.

Please note that each receiver function:

• Must be callable
• Must accept the first sender parameter of the receive sender class
• must accept the**kwargs parameter as a parameter sent in each ormar. Signals can change at any time, so your function must serve them.
• Must be asynchronous because callbacks are collected and waiting.

pre_update Currently only one parameter is sent besides the sender, and it is instance one.

Notice how the\pre_update decorator accepts the senders parameter, which can be a single model or a list of models for which you want to run the signal receiver.

Currently, signals cannot be set for all models immediately without explicitly passing them all into the receiver's registry.

import asyncio

import databases
import ormar
import sqlalchemy
from examples import create_drop_database
from ormar import pre_update

DATABASE_URL = "sqlite:///test.db"

ormar_base_config = ormar.OrmarConfig(
    database=databases.Database(DATABASE_URL),
    metadata=sqlalchemy.MetaData(),
)


class Album(ormar.Model):
    ormar_config = ormar_base_config.copy(
        tablename="albums",
    )

    id: int = ormar.Integer(primary_key=True)
    name: str = ormar.String(max_length=100)
    is_best_seller: bool = ormar.Boolean(default=False)
    play_count: int = ormar.Integer(default=0)


@pre_update(Album)
async def before_update(sender, instance, **kwargs):
    if instance.play_count > 50 and not instance.is_best_seller:
        instance.is_best_seller = True


@create_drop_database(base_config=ormar_base_config)
async def run_query():
    # here album.play_count ans is_best_seller get default values
    album = await Album.objects.create(name="Venice")
    assert not album.is_best_seller
    assert album.play_count == 0

    album.play_count = 30
    # here a trigger is called but play_count is too low
    await album.update()
    assert not album.is_best_seller

    album.play_count = 60
    await album.update()
    assert album.is_best_seller


asyncio.run(run_query())

!!!note Note that the sink is defined at class level->, so even if you connect/disconnect the function through an instance, it will run/stop all operations on that ormar.Model class.

Note that our newly created function has classes of instances and instances, so you can easily run database queries inside the sink if you like.

import asyncio

import databases
import ormar
import sqlalchemy
from examples import create_drop_database
from ormar import pre_update

DATABASE_URL = "sqlite:///test.db"

ormar_base_config = ormar.OrmarConfig(
    database=databases.Database(DATABASE_URL),
    metadata=sqlalchemy.MetaData(),
)


class Album(ormar.Model):
    ormar_config = ormar_base_config.copy(
        tablename="albums",
    )

    id: int = ormar.Integer(primary_key=True)
    name: str = ormar.String(max_length=100)
    is_best_seller: bool = ormar.Boolean(default=False)
    play_count: int = ormar.Integer(default=0)


@pre_update(Album)
async def before_update(sender, instance, **kwargs):
    if instance.play_count > 50 and not instance.is_best_seller:
        instance.is_best_seller = True


@create_drop_database(base_config=ormar_base_config)
async def run_query():
    # here album.play_count ans is_best_seller get default values
    album = await Album.objects.create(name="Venice")
    assert not album.is_best_seller
    assert album.play_count == 0

    album.play_count = 30
    # here a trigger is called but play_count is too low
    await album.update()
    assert not album.is_best_seller

    album.play_count = 60
    await album.update()
    assert album.is_best_seller


asyncio.run(run_query())

You can define the same receiver for multiple models at once by passing a list of models to the signal decorator.

# define a dummy debug function
@pre_update([Album, Track])
async def before_update(sender, instance, **kwargs):
    print(f"{sender.get_name()}: {instance.model_dump_json()}: {kwargs}")

Of course, you can also create multiple functions for the same signal and model. Each of them runs on each signal.

@pre_update(Album)
async def before_update(sender, instance, **kwargs):
    print(f"{sender.get_name()}: {instance.model_dump_json()}: {kwargs}")

@pre_update(Album)
async def before_update2(sender, instance, **kwargs):
    print(f'About to update {sender.get_name()} with pk: {instance.pk}')

Note that ormar decorators are syntactic sugar, you can directly connect functions or methods of a given signal of a given model. Connect accepts only one parameter-your receiver function/method.

class AlbumAuditor:
    def __init__(self):
        self.event_type = "ALBUM_INSTANCE"

    async def before_save(self, sender, instance, **kwargs):
        await AuditLog(
            event_type=f"{self.event_type}_SAVE", event_log=instance.model_dump_json()
        ).save()

auditor = AlbumAuditor()
pre_save(Album)(auditor.before_save)
# call above has same result like the one below
Album.ormar_config.signals.pre_save.connect(auditor.before_save)
# signals are also exposed on instance
album = Album(name='Miami')
album.signals.pre_save.connect(auditor.before_save)

Note that the signal retains references to the receiver (not weak references), so keep this in mind to avoid circular references.

Disconnect the receiver

To disconnect a receiver and stop it from running a given model, you need to disconnect it.

@pre_update(Album)
async def before_update(sender, instance, **kwargs):
    if instance.play_count > 50 and not instance.is_best_seller:
        instance.is_best_seller = True

# disconnect given function from signal for given Model
Album.ormar_config.signals.pre_save.disconnect(before_save)
# signals are also exposed on instance
album = Album(name='Miami')
album.signals.pre_save.disconnect(before_save)

available signal

!!!Warning Please note that no signal will be sent:

*  bulk operations (`QuerySet.bulk_create` and `QuerySet.bulk_update`) as they are designed for speed.

*  queryset table level operations (`QuerySet.update` and `QuerySet.delete`) as they run on the underlying tables 
(more like raw sql update/delete operations) and do not have specific instance.

pre save

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

Send the Model.save () and Model.objects.create() methods.

sender is an ormar.Model class and instance is the model to save.

after saving

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

Send Model.save () and Model.objects.create() methods.

sender is an ormar.Model class and instance is the saved model.

pre-update

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

Send to Model.update() method.

sender is an ormar.Model class and instance is the model to update.

updated

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

Send to Model.update() method.

sender is an ormar.Model class and instance is an updated model.

predeleted

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

Send the Model.save () and Model.objects.create() methods.

sender is an ormar.Model class and instance is the model to delete.

after deletion

post_delete(sender: Type["Model"], instance: "Model")

Send to Model.update() method.

sender is an ormar.Model class and instance is the model to be deleted.

Pre-relationship addition

pre_relation_add(sender: Type["Model"], instance: "Model", child: "Model", relation_name: str, passed_args: Dict)

Send Model.relation_name.add() method of many-to-many relationship and inverse of foreign key relationship.

sender -sender class, instance -add instance of related model, child -model being added, relation_name -name of relation to which child is added, also passed for add signal_kwargs -dictionary of kwargs passed to add()

Post-relationship addition

post_relation_add(sender: Type["Model"], instance: "Model", child: "Model", relation_name: str, passed_args: Dict)

Send Model.relation_name.add() method of many-to-many relationship and inverse of foreign key relationship.

sender -sender class, instance -add instance of related model, child -model being added, relation_name -name of relation to which child is added, also passed for add signal_kwargs -dictionary of kwargs passed to add()

Pre-relationship deletion

pre_relation_remove(sender: Type["Model"], instance: "Model", child: "Model", relation_name: str)

Send the Model.relation_name.remove() method of the many-to-many relationship and the inverse of the foreign key relationship.

sender -sender class, instance -add instance of related model, child -model being added, relation_name -name of relation to which child is added.

After deleting the relationship

post_relation_remove(sender: Type["Model"], instance: "Model", child: "Model", relation_name: str, passed_args: Dict)

Send the Model.relation_name.remove() method of the many-to-many relationship and the inverse of the foreign key relationship.

sender -sender class, instance -add instance of related model, child -model being added, relation_name -name of relation to which child is added.

After batch update

post_bulk_update(sender: Type["Model"], instances: List["Model"], **kwargs), sent to Model.objects.bulk_update(List[objects]) method.

Define your own signals

Note that you can create your own signals, although you have to send them manually in code or subclass ormar.Model and trigger signals there.

Creating a new signal is easy. The following example sets up a new signal named your_custom_signal.

import databases
import sqlalchemy

import ormar


base_ormar_config = ormar.OrmarConfig(
    database=databases.Database("sqlite:///db.sqlite"),
    metadata=sqlalchemy.MetaData(),
)


class Album(ormar.Model):
    ormar_config = base_ormar_config.copy()

    id: int = ormar.Integer(primary_key=True)
    name: str = ormar.String(max_length=100)
    is_best_seller: bool = ormar.Boolean(default=False)
    play_count: int = ormar.Integer(default=0)

Album.ormar_config.signals.your_custom_signal = ormar.Signal()
Album.ormar_config.signals.your_custom_signal.connect(your_receiver_name)

In fact, the bottom layer of a signal is a SignalEmitter instance that holds a dictionary of known signals and allows you to access them as attributes. SignalEmitter creates a signal for you when you try to access a signal that doesn't exist.

So the example above can be simplified to. The signal will be created for you.

Album.ormar_config.signals.your_custom_signal.connect(your_receiver_name)

Now to trigger this signal, you need to call the send method of the signal.

await Album.ormar_config.signals.your_custom_signal.send(sender=Album)

Note that sender is the only required parameter, which should be the ormar Model class.

Additional parameters must be passed as keyword parameters.

await Album.ormar_config.signals.your_custom_signal.send(sender=Album, my_param=True)