シグナル

シグナルは、ormarで特定のタイプのイベントが発生するたびに、コード（関数/メソッド）をトリガーするメカニズムです。

これを行うには、選択したモデルの特定のタイプの信号の受信機を登録する必要があります。

受信者の定義

モデルの例を次に示します。

データベースのインポート
sqlalchemyをインポート

インポートまたは


base_ormar_config=ormar. OrmarConfig(ベースの設定
    Database("sqlite:///db. sqlite"),
    metadata=sqlalchemy. MetaData(),
)


class Album(ormar. Model):
    base_or_config. copy()

    id: int=ormar.Integer(プライマリキー=True)
    name: str=ormar.String(max_length=100)
    is_best_セラー: bool=ormar.Boolean(デフォルト=False)
    play_count: int=ormar.Integer(デフォルト=0)

たとえば、アルバムが50回以上再生された場合に_best_album.is状態に設定されるトリガーを定義できます。

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つだけ送信しており、インスタンス1です。

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

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

モデルのリストをシグナルデコレータに渡すことで、複数のモデルに対して同じレシーバを一度に定義できます。

#ダミーデバッグ関数を定義する
@pre_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_update 2(sender, instance,**kwargs):
    print(f'pk:{instance.pk}で{sender.get_name()}を更新しようとしています')

ormarデコレータは、与えられたモデルの与えられたシグナルの関数またはメソッドを直接接続できるシンタックスシュガーであることに注意してください。Connectは、受信機の関数/メソッドという1つの引数のみを受け入れます。

クラスAlbumAuditor:
    def__in it__(self):
        self. event_type="ALBUM_INSTANCE"

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

Auditor=AlbumAuditor()
pre_save(アルバム)(Auditor. before_save)
#上記の呼び出しは、以下の呼び出しと同じ結果になります。
Album. or mer_config.シグナル.pre_save.connect(監査人.before_save)
#シグナルはインスタンスでも公開されます
アルバム=アルバム(name='Miami')
アルバム.シグナル. pre_save.connect(監査人.before_save)

！！！警告信号は（弱い参照ではなく）レシーバへの参照を保持するので、ループ参照を避けるためにこれを覚えておいてください。

受信機を切断します

レシーバーを切断し、特定のモデルの実行を停止するには、レシーバーを切断する必要があります。

@pre_update(アルバム)
async def before_update(sender, instance,**kwargs):
    もしinstance. play_countが50以上で、instance.is_best_sellでない場合:
        ベストセラーのインスタンス

#与えられた関数を与えられたモデルの信号から切り離す
Album. ormar_config.シグナル.pre_save.disconnect(before_save)
#シグナルはインスタンスでも公開されます
アルバム=アルバム(name='Miami')
アルバム.シグナル. pre_save.disconnect(before_save)

利用可能な信号

!!!警告信号は送信されません。

*  の一括操作（`QuerySet.bulk_create` と `QuerySet.bulk_update`）は、速度を考慮して設計されています。

*  クエリセットテーブルレベルの操作(`QuerySet.update` および `QuerySet.delete`)
（生のSQLの更新/削除操作に似ています）そして、特定のインスタンスを持っていません。

事前保存

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

Model.save()とModel. object.create()メソッドを送信します。

senderはormar. Modelクラスで、instanceは保存するモデルです。

保存後

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

Model.save()とModel. object.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. object.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. Relations_name.add()メソッドと外部キー関係の逆を送信します。

sender-送信者クラス、instance-関連するモデルを追加するインスタンス、child-追加中のモデル、Relations_name-サブアイテムを追加する関係の名前、追加信号にも渡される_kwargs-add()のkwargsに渡される辞書

関係を追加した後

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

多対多関係のModel. Relations_name.add()メソッドと外部キー関係の逆を送信します。

sender-送信者クラス、instance-関連するモデルを追加するインスタンス、child-追加中のモデル、Relations_name-サブアイテムを追加する関係の名前、追加信号にも渡される_kwargs-add()のkwargsに渡される辞書

事前リレーションシップの削除

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

多対多関係のModel. Relations_name.remove()メソッドと外部キー関係の逆を送信します。

sender-送信者クラス、instance-関連するモデルを追加するためのインスタンス、child-追加するモデル、Relations_name-サブアイテムを追加するためのリレーションシップの名前。

関係を削除した後

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

多対多関係のModel. Relations_name.remove()メソッドと外部キー関係の逆を送信します。

sender-送信者クラス、instance-関連するモデルを追加するためのインスタンス、child-追加するモデル、Relations_name-サブアイテムを追加するためのリレーションシップの名前。

バッチ更新後

post_bulk_update(sender: Type["Model"], instances: List["Model"], **kwargs)、Model.objects.bulk_update(List[objects]) メソッドに送信します。

独自の信号を定義します

コードまたはサブクラスormar. Modelでそれらを手動で送信し、そこでシグナルをトリガーする必要がありますが、独自のシグナルを作成できることに注意してください。

新しいシグナルを作成するのは非常に簡単です。次の例では、your_stom_channelという名前の新しいシグナルを設定します。

データベースのインポート
sqlalchemyをインポート

インポートまたは


base_ormar_config=ormar. OrmarConfig(ベースの設定
    Database("sqlite:///db. sqlite"),
    metadata=sqlalchemy. MetaData(),
)


class Album(ormar. Model):
    base_or_config. copy()

    id: int=ormar.Integer(プライマリキー=True)
    name: str=ormar.String(max_length=100)
    is_best_セラー: bool=ormar.Boolean(デフォルト=False)
    play_count: int=ormar.Integer(デフォルト=0)

Album. ormar_config.シグナル。あなたのカスタムシグナル=ormar.Signal()
Album. or mer_config.シグナル.your_custom_シグナル.connect(your_レシーバー名)

実際には、シグナルの下層には既知のシグナルのディクショナリを保持し、それらをプロパティとしてアクセスできるSignalEmitterインスタンスがあります。存在しないシグナルにアクセスしようとすると、SignalEmitterがシグナルを作成します。

ということで上記の例は簡略化できます。シグナルが作成されます。

Album. or mer_config.シグナル.your_custom_シグナル.connect(your_レシーバー名)

このシグナルをトリガーするには、シグナルの送信メソッドを呼び出す必要があります。

await Album. ormer_config.シグナル.your_custom_シグナル.send(sender=Album)

senderは唯一の必須パラメータであり、ormar Modelクラスである必要があることに注意してください。

追加引数はキーワード引数として渡す必要があります。

await Album. order_config.シグナル.your_custom_シグナル.send(sender=Album,my_param=True)