Сигнал

Сигнал-это механизм, который запускает ваш код (функцию/метод) каждый раз, когда происходит событие определенного типа в ormar.

Для этого вам необходимо зарегистрировать приемник сигнала определенного типа для выбранной модели.

Определить получателя

Приведем следующие примеры моделей:

Import Data abases
sqimport alchemy

Import ormar


«Органа»((
    Database=databases.Database("sqlite:///db.sqlite")
    metadata= sqalchemy (),
)


Class Album(Ormar):
    ormar_config = base_ormar_config ()

    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)

Например, можно определить триггер, который, если альбом воспроизводится более 50 раз, устанавливает состояние album.is_best_seller.

Импортируйте декор pre_update, чтобы просмотреть список доступных в настоящее время декораторов/сигналов ниже.

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())

Определите свои функции.

Обратите внимание, что функции каждого приемника:

• Он должен быть доступным.
• Необходимо принять параметр первого отправителя для получения класса отправленного объекта
• должен принимать параметр**kwargs в качестве параметра, отправляемого в каждый ormar. Сигналы могут быть изменены в любое время, поэтому ваша функция должна обслуживать их.
• Он должен быть асинхронным, так как обратный вызов будет собираться и ждать.

pre_update в настоящее время отправляет только один аргумент, кроме отправителя, и это экземпляр 1.

Обратите внимание на то, как декор pre_update принимает параметр senders, который может представлять собой одну модель или список моделей, для которых необходимо запустить приемник сигнала.

В настоящее время невозможно сразу задать сигналы для всех моделей, не обеспечив явно передачу всех моделей в реестр приемника.

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 Обратите внимание, что приемник определен на уровне класса->, поэтому он будет запускать/останавливать все операции на этом классе ormar.Model даже при помощи функции подключения/отключения экземпляра.

Обратите внимание, что наши недавно созданные функции имеют экземпляры и классы экземпляров, поэтому вы можете легко запустить запрос базы данных в приемнике, если хотите.

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 a dummy debug function
re_update ([альбом, трек])
async def before_update (sender, instance, **kwargs):
    print(f"{sender.get_name()}:{instance.model_dump_json()}:{kwargs}")

Конечно, вы также можете создать несколько функций для одного и того же сигнала и модели. Каждый из них работает при каждом сигнале.

pre_update (альбом)
async def before_update (sender, instance, **kwargs):
    print(f"{sender.get_name()}:{instance.model_dump_json()}:{kwargs}")

pre_update (альбом)
async def before_update2(sender, instance, **kwargs):
    print(f 'About to update{sender.get_name()}with pk:{instance.pk}')

Обратите внимание, что ормар представляет собой синтаксический сахар, который вы можете напрямую подключить функцию или метод данного сигнала к данной модели. Connect принимает только один аргумент-функцию/метод вашего приемника.

Class AlbumAuditor:
    def_init__(self):
        self. event_type = "ALBUM_INSTANCE"

    async def before_save (self, sender, instance, **kwargs):
        Аварийная система (
            event_type =f"{self.event_type}_SAVE", event_log = instance.model_dump_json ()
        save()

Auditor (AlbumAuditor)
pre_save (альбом)(auditor.before_save)
Call Above has same result like the one below
Album.ormar_config signals.pre_save.connect (auditor.before_save)
# signals are and exposed on instance
Альбом (name ='Miami')
Альбом: signals.pre_save.connect (auditor.before_save)

!!! Обратите внимание, что сигнал сохраняет ссылки на приемник (а не слабые ссылки), поэтому помните об этом, чтобы избежать циклических ссылок.

Отключить приемник

Чтобы отключить приемник и остановить запуск данной модели, необходимо отключить его.

pre_update (альбом)
async def before_update (sender, instance, **kwargs):
    if instance.play> 50 и not instance.is:
        instance.is = True

# disconnect given function from signal for given Model
Album.ormar_config signals.pre_save.disconnect (before_save)
# signals are and exposed on instance
Альбом (name ='Miami')
Альбом: signals.pre_save.disconnect (before_save)

Доступные сигналы

!!!Обратите внимание, что сигнал не передается:

*  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(sender: Type["Model"], instance: "Model")

Отправьте методы Model.save() и Model.objects.create ().

sender-это класс ormar.Model, а instance-модель, которую нужно сохранить.

После сохранения

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

Отправьте методы Model.save() и Model.objects.create ().

sender-это класс ormar.Model, а instance-сохраненная модель.

Предварительное обновление

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

Отправить в метод Model.update ().

Sender-это модель ormar.Model, а instance-модель, которую нужно обновить.

После обновления

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

Отправить в метод Model.update ().

Sender-это класс ormar.Model, а instance-обновленная модель.

Предварительное удаление

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

Отправьте методы Model.save() и Model.objects.create ().

sender-это класс ormar.Model, а instance-модель, которую нужно удалить.

После удаления

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

Отправить в метод Model.update ().

sender-это класс ormar.Model, а instance-удаленная модель.

Пресвязывание Добавить

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

Отправка метода множественно-многопрофильной связи Model.relation_name.add () и обратной связи внешних ключей.

sender -класс отправителя, instance -добавление экземпляра связанной модели, child -добавляемая модель, relation_name-имя связи, добавляемой дочерними элементами, также передается_kwargs-словарь kwargs, передаваемый в add()

Добавить связь после

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

Отправка метода множественно-многопрофильной связи Model.relation_name.add () и обратной связи внешних ключей.

sender -класс отправителя, instance -добавление экземпляра связанной модели, child -добавляемая модель, relation_name-имя связи, добавляемой дочерними элементами, также передается_kwargs-словарь kwargs, передаваемый в add()

Предварительное удаление

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

Отправка метода множественно-многопрофильного отношения Model.relation_name.remove () и обратной связи внешних ключей.

sender -класс отправителя, instance -добавление экземпляра связанной модели, child -добавляемая модель, relation_name-имя связи, к которой добавляется дочерний элемент.

После удаления связи

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

Отправка метода множественно-многопрофильного отношения Model.relation_name.remove () и обратной связи внешних ключей.

sender -класс отправителя, instance -добавление экземпляра связанной модели, child -добавляемая модель, relation_name-имя связи, к которой добавляется дочерний элемент.

После пакетного обновления

post_bulk_update(sender: Type["Model"], instances: List["Model"], **kwargs), отправляйте в метод Model.objects.bulk_update(List[objects]).

Определите свой собственный сигнал.

Обратите внимание, что вы можете создавать свои собственные сигналы, хотя вам придется отправлять их вручную в коде или подклассе ormar.Model и запускать сигналы там.

Создать новый сигнал очень просто. В следующем примере устанавливается новый сигнал с именем your_custom_signal.

Import Data abases
Sqimport alchemy

import ormar


«Органа»((
    Database=databases.Database("sqlite:///db.sqlite")
    metadata= sqalchemy (),
)


Class Album(Ormar):
    ormar_config = base_ormar_config ()

    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_resteiver_name)

На самом деле, нижний слой сигнала представляет собой экземпляр SignalEmitter, который сохраняет словарь известных сигналов и позволяет вам получить к ним доступ в качестве атрибутов. Когда вы пытаетесь получить доступ к сигналу, который не существует, SignalEmitter создаст сигнал для вас.

Таким образом, приведенный выше пример можно упростить, так как будет создан сигнал для вас.

Album.ormar_config signals. your_custom_signal.connect (your_resteiver_name)

Теперь, чтобы запустить этот сигнал, необходимо вызвать метод передачи сигнала.

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

Обратите внимание, что sender является единственным требуемым параметром, и это должен быть класс ormar Model.

Дополнительные параметры должны передаваться как ключевые параметры.

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