신호

신호는 특정 유형의 이벤트가 ormar에서 발생할 때마다 코드(함수/방법)를 트리거하는 메커니즘입니다.

따라서 선택한 모델의 지정된 신호 유형에 대해 수신기를 등록해야 합니다.

수취인 정의

다음 예제 모델이 제공됩니다.

데이터베이스 가져오기
sqlalchemy 가져오기

오마르 가져오기


base_ormar_config = ormar. OrmarConfig (
    데이터베이스 = 데이터베이스. 데이터베이스("sqlite://db.sqlite"),
    메타데이터 = sqlalchemy. 메타데이터(),
)


클래스 앨범(ormar. 모델):
    ormar_config = base_ormar_config 사본()

    ID: int = ormar. 정수(prial_key = 참)
    이름: str = ormar. 문자열(max_long = 100)
    is_est_: bool = ormar. 부울(기본값 = 거짓)
    플레이 카운트: int = ormar. 정수(기본값 = 0)

예를 들어 앨범이 50번 이상 재생되는 경우 album.is_best_ 상태를 설정하는 트리거를 정의할 수 있습니다.

사전_업데이트 데코레이터를 가져오고 아래에서 현재 사용 가능한 데코레이터/신호 목록을 참조하십시오.

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_는 현재 보낸 사람 외에 하나의 매개 변수만 전송하며 인스턴스 1입니다.

_ 데코레이터가 신호 수신기를 실행할 단일 모델 또는 모델 목록이 될 수 있는 발신자 매개 변수를 어떻게 수락하는지 확인하십시오.

현재 모든 모델을 수신자의 레지스트리에 명시적으로 전달하지 않고는 모든 모델에 대한 신호를 즉시 설정할 수 없습니다.

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

!!!수신기는 클래스 레벨 ->에서 정의되므로 인스턴스를 통해 함수를 연결/연결 해제하더라도 해당 ormar에서 모든 작업을 실행/중지합니다. 모델 클래스.

새로 만든 기능에는 인스턴스 및 클래스가 있으므로 원하는 경우 싱크대 내에서 데이터베이스 쿼리를 쉽게 실행할 수 있습니다.

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_ ([앨범, 트랙])
Async def _(발신자, 인스턴스, **kwargs):
    인쇄(f "{sender.get_name()}:{instance.model_dump_json()}:{kwargs}")

물론 동일한 신호와 모델에 대해 여러 기능을 만들 수도 있습니다. 각각의 신호는 각 신호에서 실행됩니다.

@pre_ (앨범)
Async def _(발신자, 인스턴스, **kwargs):
    인쇄 (f "{sender.get_name()}:{instance.model_dump_json()}:{kwargs}")

@pre_ (앨범)
Async def _update2(발신자, 인스턴스, **kwargs):
    인쇄(f 'pk로{sender.get_name()}를 업데이트하려고 합니다:{instance.pk}')

ormar ator는 통사적 설탕이며 특정 모델에 대해 주어진 신호의 함수 또는 방법을 직접 연결할 수 있습니다. 연결에는 수신기 기능/방법이라는 하나의 인수만 필요합니다.

클래스 앨범 감사자:
    데프__init__(자체):
        Self.event_type="ALBUM_INSTANCE"

    Async def _Save(자체, 발신자, 인스턴스, **kwargs):
        감사 로그 대기 중(
            event_type = f "{self.event_type}_SAVE", event_log = instance. model_dump_json ()
        ).저장 ()

감사관 = 앨범감사관()
pre_Save (앨범) (감사. beat_Save)
# 위의 통화는 아래의 통화와 동일한 결과를 가집니다.
ormar_config 신호 pre_Save Connect(감사자 bebebe_Save)
# 신호는 인스턴스에도 노출됩니다.
앨범 = 앨범 (이름 = 'Miami')
Album . 시그널 . pre_Save 연결 (감사 . beat_Save)

!!! 경고 신호는 약한 참조가 아닌 수신기에 대한 참조를 유지하므로 원형 참조를 피하도록 주의하십시오.

수신기 분리

수신기를 분리하고 지정된 모델을 실행하지 못하도록 하려면 수신기를 분리해야 합니다.

@pre_ (앨범)
Async def _(발신자, 인스턴스, **kwargs):
    인스턴스가 아닌 경우. play_count. 50
        인스턴스. is_best _ = 참

# 주어진 모델에 대한 신호에서 주어진 기능을 분리합니다.
Album. ormar_config 신호. pre_Save (beat_Save)
# 신호는 인스턴스에도 노출됩니다.
앨범 = 앨범 (이름 = 'Miami')
Alum.Sigals.pre_Save 연결 끊기 (bebeat_Save)

사용 가능한 신호

!!!경고, 신호가 전송되지 않습니다.

*  대량 작업(`QuerySet.bulk_create` 및 `QuerySet.bulk_update`)은 속도를 위해 설계되었습니다.

*  쿼리 테이블 수준 작업 설정(`QuerySet.update` 그리고 `QuerySet.delete`) 기본 테이블에서 실행할 때 
(원시 sql 업데이트/삭제 작업과 유사) 및 특정 인스턴스가 없습니다.

사전 저장

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

Model.save() 및 Model.ts.create() 메서드를 보냅니다.

보낸 사람은 ormar입니다. 모델 클래스, 저장할 모델의 인스턴스입니다.

저장 후

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

Model.save() 및 Model.ts.create() 메서드를 보냅니다.

보낸 사람은 ormar입니다. 모델 클래스, 저장된 모델의 인스턴스입니다.

사전 업데이트

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

Model.() 메서드로 전송되었습니다.

보낸 사람은 ormar입니다. 모델 클래스, 업데이트 할 모델의 인스턴스입니다.

업데이트 후

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

Model.() 메서드로 전송되었습니다.

보낸 사람은 ormar입니다. 모델 클래스, 업데이트된 모델의 인스턴스입니다.

사전 삭제

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

Model.save() 및 Model.ts.create() 메서드를 보냅니다.

보낸 사람은 ormar입니다. 모델 클래스, 삭제할 모델의 인스턴스입니다.

삭제 후

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

Model.() 메서드로 전송되었습니다.

보낸 사람은 ormar입니다. 모델 클래스, 삭제되는 모델의 인스턴스입니다.

관계 전 추가

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

다대다 관계의 Model._name() 메서드와 외래 키 관계의 역방향을 보냅니다.

보낸 사람 - 보낸 사람 클래스, 인스턴스 - 관련 모델의 인스턴스 추가, 하위 - 모델 추가, 관계 \ _ 이름 - 하위 추가를 위한 관계 이름, 추가 신호 \ _ kwargs - 추가하기 위해 전달된 kwargs 사전 ()

포스트 관계 추가

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

다대다 관계의 Model._name() 메서드와 외래 키 관계의 역방향을 보냅니다.

보낸 사람 - 보낸 사람 클래스, 인스턴스 - 관련 모델의 인스턴스 추가, 하위 - 모델 추가, 관계 \ _ 이름 - 하위 추가를 위한 관계 이름, 추가 신호 \ _ kwargs - 추가하기 위해 전달된 kwargs 사전 ()

관계 전 삭제

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

다대다 관계의 Model._name() 메서드와 외래 키 관계의 역방향을 보냅니다.

보낸 사람 - 보낸 사람 클래스, 인스턴스 - 관련 모델의 인스턴스 추가, 하위 - 모델 추가, _name - 하위를 추가할 관계의 이름입니다.

관계를 삭제한 후

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

다대다 관계의 Model._name() 메서드와 외래 키 관계의 역방향을 보냅니다.

보낸 사람 - 보낸 사람 클래스, 인스턴스 - 관련 모델의 인스턴스 추가, 하위 - 모델 추가, _name - 하위를 추가할 관계의 이름입니다.

일괄 업데이트 후

post_bulk_update(sender: Type["Model"], instances: List["Model"], **kwargs) , Model.objects.bulk_update(List[objects]) 방법.

자신의 신호를 정의하십시오.

코드 또는 하위 클래스 오마로 수동으로 보내야 하지만 자신만의 신호를 만들 수 있습니다. 신호를 모델링하고 트리거합니다.

새 신호를 만드는 것은 간단합니다. 다음 예에서는 __ 신호라는 새 신호를 설정합니다.

데이터베이스 가져오기
sqlalchemy 가져오기

오마르 가져오기


base_ormar_config = ormar. OrmarConfig (
    데이터베이스 = 데이터베이스. 데이터베이스("sqlite://db.sqlite"),
    메타데이터 = sqlalchemy. 메타데이터(),
)


클래스 앨범(ormar. 모델):
    ormar_config = base_ormar_config 사본()

    ID: int = ormar. 정수(prial_key = 참)
    이름: str = ormar. 문자열(max_long = 100)
    is_est_: bool = ormar. 부울(기본값 = 거짓)
    플레이 카운트: int = ormar. 정수(기본값 = 0)

Album. ormar_config 신호. 당신의 _t_=ormar. 신호()
ormar_config는 당신의 __ 연결을 신호합니다 (당신의 _ver_name)

사실, 신호의 기본 계층은 알려진 신호의 사전을 보관하고 속성으로 액세스할 수 있는 신호 송신기 인스턴스입니다. 존재하지 않는 신호에 액세스하려고 하면 신호 송신기가 신호를 생성합니다.

따라서 위의 예는 다음과 같이 단순화할 수 있습니다. 신호가 생성됩니다.

1	`ormar_config는 당신의 __ 연결을 신호합니다 (당신의 _ver_name)`

이제 이 신호를 트리거하려면 신호의 송신 방법을 호출해야 합니다.

Album. ormar_config 신호를 기다립니다. 당신의 __ 보내기 (발신자 = 앨범)

보낸 사람은 필요한 유일한 매개 변수이며 Ormar Model 클래스여야 합니다.

추가 매개 변수는 키워드 인수로 전달되어야 합니다.

Album. ormar_config 신호를 기다립니다. 당신의 __ 보내기 (발신자 = 앨범, my_param = 참)