Le signal

Les signaux sont un mécanisme qui déclenche votre code (fonction/méthode) chaque fois qu 'un type donné d'événement se produit dans ormar.

Pour ce faire, vous devez enregistrer un récepteur pour un type de signal donné du modèle sélectionné.

Définir le destinataire

Un exemple de modèle est donné comme suit:

Importer des databases
Import sqlalchemy

Import par Ormar


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


Classe Album(modèle ormar):
    ormar_config = base_ormar_config.copy()

    id: int = ormar.Integer(primary_key=True)
    nom: str = ormar.String(max_length=100)
    is_best_sell: bool = ormar. booléen (default=False)
    play_count: int = ormar.Integer(default=0)

Par exemple, vous pouvez définir un déclencheur qui définira le statut album.is_best_sell si l'album a été lu plus de 50 fois.

Importez le décorateur pre_update pour voir la liste des décorateurs/signaux actuellement disponibles ci-dessous.

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

Définissez votre fonction.

Veuillez noter que chaque récepteur fonctionne:

• Doit être invocable
• DOIT accepter le premier paramètre de l'expéditeur de la classe d'objet envoyé
• doit accepter le paramètre**kwargs comme paramètre envoyé dans chaque paramètre ormar. les signaux peuvent changer à tout moment, votre fonction doit donc les servir.
• Il doit être asynchrone, car les callbacks sont collectés et attendent.

pre_update n'envoie actuellement qu 'un seul argument en plus de l'expéditeur, et c'est l'instance un.

Remarquez comment le décorateur pre_update accepte le paramètre senders, qui peut être un modèle unique ou une liste de modèles pour lesquels vous souhaitez exécuter le récepteur de signal.

Actuellement, il n'est pas possible de définir immédiatement un signal pour tous les modèles sans passer explicitement tous les modèles dans l'enregistrement du récepteur.

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

!!!Notez que le récepteur est défini au niveau de la classe->, donc même si vous connectez/déconnectez une fonction via une instance, il exécutera/arrêtera toutes les opérations sur cette classe. modèle.

Notez que nos fonctions nouvellement créées ont des instances et des classes pour les instances, vous pouvez donc facilement exécuter des requêtes de base de données à l'intérieur du récepteur si vous le souhaitez.

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

Vous pouvez définir le même récepteur pour plusieurs modèles à la fois en passant la liste des modèles au décorateur de signal.

#définir une fonction de débogage dummy
@pre_update([Album, piste])
async def before_update(sender, instance, **kwargs):
    print(f"{sender.get_name()}:{instance.model_dump_json()}:{kwargs}")

Bien sûr, vous pouvez également créer plusieurs fonctions pour le même signal et le même modèle-chacune d'elles sera exécutée à chaque 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}')

Notez que les décorateurs ormar sont des sucres de syntaxe, et que vous pouvez connecter directement des fonctions ou des méthodes pour un signal donné d'un modèle donné. Connect n'accepte qu 'un seul paramètre-votre fonction/méthode de récepteur.

Classe 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 a eu le même résultat que celui que l'on a vu
Album.ormar_config.signals.pre_save.connect(auditor.before_save)
#signaux sont aussi exposés sur l'instance
album = Album(nom ="Miami")
album.signals.pre_save.connect(auditor.before_save)

Attention!!! Notez que le signal conserve une référence au récepteur (et non une référence faible), alors gardez cela à l'esprit pour éviter les références circulaires.

Déconnecter le récepteur

Pour déconnecter le récepteur et l'empêcher de fonctionner sur un modèle donné, vous devez le déconnecter.

@pre_update(Album)
async def before_update(sender, instance, **kwargs):
    Si instance.play_count > 50 et pas instance.is_best_sell:
        instance.is_best_sell = True

#déconnecter la fonction de don du signal pour le modèle donné
Album.ormar_config.signals.pre_save.disconnect(before_save)
#signaux sont aussi exposés sur l'instance
album = Album(nom ="Miami")
album.signals.pre_save.disconnect(avant_save)

Signaux disponibles

!!!Attention Veuillez noter qu 'aucun signal n'est envoyé:

*  bulk operations (`QuerySet.bulk_create` et `QuerySet.bulk_update`) comme ils sont conçus pour la vitesse.

*  queryset table level operations (`QuerySet.update` et `QuerySet.delete`) comme ils courent sur les tables underlying
(more comme raw sql update/delete operations) et ne pas avoir d'instance spécifique.

Pré-sauvegarde

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

Envoyez les méthodes Model.save() et Model.objects.create().

sender est une classe ormar.Model et instance est le modèle à sauvegarder.

Après conservation

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

Envoyez les méthodes Model.save() et Model.objects.create().

Sender est une classe ormar.Model et instance est le modèle enregistré.

Pré-actualisation

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

Envoyé à la méthode Model.update().

Sender est une classe ormar.Model et instance est le modèle à mettre à jour.

Après la mise à jour

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

Envoyé à la méthode Model.update().

Sender est une classe ormar.Model et instance est un modèle mis à jour.

Pré-supprimé

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

Envoyez les méthodes Model.save() et Model.objects.create().

sender est une classe ormar.Model et instance est le modèle à supprimer.

Après suppression

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

Envoyé à la méthode Model.update().

Sender est une classe ormar.Model et instance est le modèle supprimé.

Pré-relation Ajouter

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

Envoie la méthode Model.relation_name.add() de la relation plusieurs-à-plusieurs et l'inverse de la relation de clé étrangère.

sender -classe d'expéditeur, instance -ajouter une instance du modèle associé, child -le modèle ajouté, relation_name -le nom de la relation dans laquelle les enfants sont ajoutés, pour ajouter un signal passe également_kwargs -le dictionnaire des kwargs passés à add()

Post-relation Ajouter

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

Envoie la méthode Model.relation_name.add() de la relation plusieurs-à-plusieurs et l'inverse de la relation de clé étrangère.

sender -classe d'expéditeur, instance -ajouter une instance du modèle associé, child -le modèle ajouté, relation_name -le nom de la relation dans laquelle les enfants sont ajoutés, pour ajouter un signal passe également_kwargs -le dictionnaire des kwargs passés à add()

Suppression de pré-relation

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

Envoie la méthode Model.relation_name.remove() de la relation plusieurs-à-plusieurs et l'inverse de la relation de clé étrangère.

sender -classe expéditeur, instance -ajouter une instance du modèle associé, child -modèle ajouté, relation_name -nom de la relation dans laquelle l'enfant est ajouté.

Après la suppression de la relation

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

Envoie la méthode Model.relation_name.remove() de la relation plusieurs-à-plusieurs et l'inverse de la relation de clé étrangère.

sender -classe expéditeur, instance -ajouter une instance du modèle associé, child -modèle ajouté, relation_name -nom de la relation dans laquelle l'enfant est ajouté.

Après une mise à jour en masse

post_bulk_update(sender: Type["Model"], instances: List["Model"], **kwargs), envoyé à la méthode Model.objects.bulk_update(List[objects]).

Définissez votre propre signal

Notez que vous pouvez créer vos propres signaux, bien que vous deviez les envoyer manuellement dans le code ou la sous-classe ormar.Model et y déclencher le signal.

La création d'un nouveau signal est très simple. l'exemple suivant définira un nouveau signal appelé votre_signal personnalisé_signal.

Importer des databases
Import sqlalchemy

Importer à Ormar


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


Classe Album(modèle ormar):
    ormar_config = base_ormar_config.copy()

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

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

En fait, la couche inférieure du signal est une instance de SignalEmitter qui contient un dictionnaire de signaux connus et vous permet d'y accéder en tant que propriété. SignalEmitter créera un signal pour vous lorsque vous essayez d'accéder à un signal qui n'existe pas.

Donc, l'exemple ci-dessus peut être simplifié à. va créer un signal pour vous.

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

Maintenant, pour déclencher ce signal, vous devez appeler la méthode d'envoi du signal.

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

Notez que sender est le seul paramètre requis et qu 'il devrait s'agir de la classe modèle ormar.

Les arguments supplémentaires doivent être passés comme arguments de mots-clés.

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