Sinais

Por vezes, pode ser necessário ouvir um evento de documento ao salvar, ou seja, deseja-se realizar uma ação específica quando algo acontece nos modelos.

O Django, por exemplo, possui esse mecanismo chamado Sinais, que pode ser muito útil para esses casos e para realizar operações extras quando uma ação ocorre no documento.

Outros ORMs adotaram uma abordagem semelhante a essa e uma excelente foi o Ormar, que adotou a abordagem do Django na sua própria implementação.

O Mongoz, sendo como é desenhado, inspirou-se em ambas as abordagens e também suporta o Sinal.

O que são Sinais

Sinais são mecanismos usados para acionar ações específicas quando ocorre um determinado tipo de evento nos modelos do Mongoz.

Da mesma forma que o Django aborda os sinais em termos de registro, o Mongoz faz isso de maneira semelhante.

Sinais padrão

O Mongoz possui receptores padrão para cada documento criado no ecossistema. Eles podem ser usados prontamente a qualquer momento.

Também existem sinais personalizados caso queira algo "extra" além dos padrões fornecidos.

Como usá-los

Os sinais estão dentro de mongoz.core.signals e para importá-los, basta executar:

from mongoz.core.signals import (
    post_delete,
    post_save,
    post_update,
    pre_delete,
    pre_save,
    pre_update,
)

pre_save

O pre_save é usado quando um documento está prestes a ser guardado e é acionado nas funções Document.save() e Document.objects.create.

pre_save(send: Type["Document"], instance: "Document")

post_save

O post_save é usado após o documento já ter sido criado e guardado na base de dados, ou seja, quando uma instância já existe após o save. Esse sinal é acionado nas funções Document.save() e Document.objects.create.

post_save(send: Type["Document"], instance: "Document")

pre_update

O pre_update é usado quando um documento está prestes a receber as atualizações e é acionado nas funções Document.update() e Document.objects.update.

pre_update(send: Type["Document"], instance: "Document")

post_update

O post_update é usado quando um documento já realizou as atualizações e é acionado nas funções Document.update() e Document.objects.update.

post_update(send: Type["Document"], instance: "Document")

pre_delete

O pre_delete é usado quando um documento está prestes a ser excluído e é acionado nas funções Document.delete() e Document.objects.delete.

pre_delete(send: Type["Document"], instance: "Document")

post_delete

O post_update é usado quando um documento já foi excluído e é acionado nas funções Document.delete() e Document.objects.delete.

post_update(send: Type["Document"], instance: "Document")

Receptor

O receptor é a função ou ação que você deseja executar quando um sinal é acionado, noutras palavras, é o que está a escutar um determinado evento.

Vejamos um exemplo. Dado o seguinte documento.

import asyncio

import mongoz

database_uri = "mongodb://localhost:27017"
registry = mongoz.Registry(database_uri)


class User(mongoz.Document):
    name: str = mongoz.String(max_length=255)
    email: str = mongoz.Email(max_length=255)
    is_verified: bool = mongoz.Boolean(default=False)

    class Meta:
        registry = registry
        database = "my_db"

Pode definir um trigger para enviar um e-mail ao utilizador registrado após a criação do registro usando o sinal post_save. A razão para usar o post_save é porque a notificação deve ser enviada após a criação do registro e não antes. Se fosse antes, o pre_save seria o sinal a ser usado.

from mongoz.core.signals import post_save


def send_notification(email: str) -> None:
    """
    Sends a notification to the user
    """
    send_email_confirmation(email)


@post_save(User)
async def after_creation(sender, instance, **kwargs):
    """
    Sends a notification to the user
    """
    send_notification(instance.email)

Como pode ver, o decorador post_save está a apontar para o documento User, ou seja, está "a escutar" eventos nesse mesmo documento.

Isso é chamado de receptor.

Pode usar qualquer um dos sinais padrão disponíveis ou até mesmo criar seu próprio sinal personalizado.

Requisitos

Ao definir a função ou receptor, ela deve atender aos seguintes requisitos:

• Deve ser um callable.
• Deve ter o argumento sender como primeiro parâmetro, que corresponde ao documento do objeto de envio.
• Deve ter o argumento **kwargs como parâmetro, pois cada documento pode mudar a qualquer momento.
• Deve ser async porque as operações de documento do Mongoz são aguardadas.

Múltiplos receptores

E se você quiser usar o mesmo receptor para vários modelos? Vamos adicionar agora um documento adicional chamado Profile.

import asyncio

import mongoz

database_uri = "mongodb://localhost:27017"
registry = mongoz.Registry(database_uri)


class User(mongoz.Document):
    name: str = mongoz.String(max_length=255)
    email: str = mongoz.Email(max_length=255)
    is_verified: bool = mongoz.Boolean(default=False)

    class Meta:
        registry = registry
        database = "my_db"


class Profile(mongoz.Document):
    profile_type: str = mongoz.String(max_length=255)

    class Meta:
        registry = registry
        database = "my_db"

A maneira de definir o receptor para ambos pode ser facilmente alcançada da seguinte forma:

from mongoz.core.signals import post_save


def send_notification(email: str) -> None:
    """
    Sends a notification to the user
    """
    send_email_confirmation(email)


@post_save([User, Profile])
async def after_creation(sender, instance, **kwargs):
    """
    Sends a notification to the user
    """
    if isinstance(instance, User):
        send_notification(instance.email)
    else:
        # something else for Profile
        ...

Desta forma, pode corresponder e executar qualquer lógica personalizada sem precisar se repetir muito e mantendo seu código limpo e consistente.

Múltiplos receptores para o mesmo documento

E se agora quiser ter mais de um receptor para o mesmo documento? Na prática, colocaria todos num só lugar, mas talvez queira fazer algo completamente diferente e dividi-los em vários.

Pode facilmente fazer isso desta forma:

from mongoz.core.signals import post_save


def push_notification(email: str) -> None:
    # Sends a push notification
    ...


def send_email(email: str) -> None:
    # Sends an email
    ...


@post_save(User)
async def after_creation(sender, instance, **kwargs):
    """
    Sends a notification to the user
    """
    send_email(instance.email)


@post_save(User)
async def do_something_else(sender, instance, **kwargs):
    """
    Sends a notification to the user
    """
    push_notification(instance.email)

Isto garantirá que cada receptor execute a ação definida.

Desconectando receptores

Se deseja desconectar o receptor e impedi-lo de ser executado para um determinado documento, também pode fazer isso de maneira simples.

from mongoz.core.signals import post_save


def send_notification(email: str) -> None:
    """
    Sends a notification to the user
    """
    send_email_confirmation(email)


@post_save(User)
async def after_creation(sender, instance, **kwargs):
    """
    Sends a notification to the user
    """
    send_notification(instance.email)


# Disconnect the given function
User.meta.signals.post_save.disconnect(after_creation)

# Signals are also exposed via instance
user.signals.post_save.disconnect(after_creation)

Sinais Personalizados

Aqui é onde as coisas ficam interessantes. Muitas vezes, você querer ter o seu próprio Sinal e não depender apenas dos padrões fornecidos, e isso é perfeitamente natural e comum.

O Mongoz permite que os sinais personalizados sejam usados de acordo com o seu próprio design.

Vamos continuar com o mesmo exemplo do documento User.

import asyncio

import mongoz

database_uri = "mongodb://localhost:27017"
registry = mongoz.Registry(database_uri)


class User(mongoz.Document):
    name: str = mongoz.String(max_length=255)
    email: str = mongoz.Email(max_length=255)
    is_verified: bool = mongoz.Boolean(default=False)

    class Meta:
        registry = registry
        database = "my_db"

Agora deseja ter um sinal personalizado chamado on_verify especificamente adaptado às necessidades e lógica do seu User.

Para defini-lo, pode simplesmente fazer:

import asyncio

import mongoz

database_uri = "mongodb://localhost:27017"
registry = mongoz.Registry(database_uri)


class User(mongoz.Document):
    name: str = mongoz.String(max_length=255)
    email: str = mongoz.Email(max_length=255)
    is_verified: bool = mongoz.Boolean(default=False)

    class Meta:
        registry = registry
        database = "my_db"


# Create the custom signal
User.meta.signals.on_verify = mongoz.Signal()

Sim, é assim simples. Só precisa adicionar um novo sinal on_verify aos sinais do documento e, a partir de agora, o documento User terá um novo sinal pronto para ser usado.

Warning

Tenha em mente que os sinais são do tipo nível de classe, o que significa que afetarão todas as instâncias derivadas dele. Esteja atento ao criar um sinal personalizado e os seus impactos.

Agora deseja criar uma funcionalidade personalizada para ser ouvida no novo Sinal.

import asyncio

import mongoz

database_uri = "mongodb://localhost:27017"
registry = mongoz.Registry(database_uri)


class User(mongoz.Document):
    name: str = mongoz.String(max_length=255)
    email: str = mongoz.Email(max_length=255)
    is_verified: bool = mongoz.Boolean(default=False)

    class Meta:
        registry = registry
        database = "my_db"


# Create the custom signal
User.meta.signals.on_verify = mongoz.Signal()


# Create the receiver
async def trigger_notifications(sender, instance, **kwargs):
    """
    Sends email and push notification
    """
    send_email(instance.email)
    send_push_notification(instance.email)


# Register the receiver into the new Signal.
User.meta.signals.on_verify.connect(trigger_notifications)

Agora, não apenas criou o novo receptor trigger_notifications, mas também o conectou ao novo sinal on_verify.

Como usá-lo

Agora é hora de usar o sinal numa lógica personalizada, afinal, ele foi criado para garantir que seja personalizado o suficiente para as necessidades da lógica de negócio.

Para simplificação, o exemplo abaixo será uma lógica muito simples.

async def create_user(**kwargs):
    """
    Creates a user
    """
    await User.query.create(**kwargs)


async def is_verified_user(id: str):
    """
    Checks if user is verified and sends notification
    if true.
    """
    user = await User.get_document_by_id(id))

    if user.is_verified:
        # triggers the custom signal
        await User.meta.signals.on_verify.send(sender=User, instance=user)

Como pode ver, o on_verify é acionado apenas se o utilizador estiver verificado e não em nenhum outro lugar.

Desconectar o sinal

O processo de desconectar o sinal é exatamente o mesmo que antes.

async def trigger_notifications(sender, instance, **kwargs):
    """
    Sends email and push notification
    """
    send_email(instance.email)
    send_push_notification(instance.email)


# Disconnect the given function
User.meta.signals.on_verify.disconnect(trigger_notifications)