Managers

Os gestores são uma ótima ferramenta que o Mongoz oferece. Fortemente inspirados pelo Edgy, os gestores permitem que construa pesquisas personalizadas prontas para serem usadas pelos documentos.

O Mongoz por defeito usa o gestor chamado objects, o que torna simples de perceber.

Vamos ver um exemplo.

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.String(max_length=70)
    is_active: bool = mongoz.Boolean(default=True)

    class Meta:
        registry = registry
        database = "my_db"


await User.objects.create(name="Mongoz", email="foo@bar.com")  # noqa

user = await User.objects.get(name="Mongoz")  # noqa
# User(id=ObjectId(...))

Ao consultar a tabela User, o objects (gestor) é o padrão e deve estar sempre presente ao fazê-lo.

Danger

Isto aplica-se apenas ao lado do manager do Mongoz, por exemplo, ao usar Document.objects..... Ao usar Document.query... isto não funcionará.

Gestor personalizado

Também é possível ter os próprios gestores personalizados e para fazer isso, deve herdar a classe QuerySetManager e substituir o get_queryset().

Para aqueles familiarizados com os gestores do Django, o princípio é exatamente o mesmo. 😀

Os gestores devem ser anotados como ClassVar ou um erro será lançado.

from typing import ClassVar

import mongoz
from mongoz import QuerySetManager

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


class InactiveManager(QuerySetManager):
    """
    Custom manager that will return only active users
    """

    def get_queryset(self) -> "QuerySetManager":
        queryset = super().get_queryset().filter(is_active=False)
        return queryset


class User(mongoz.Document):
    # Add the new manager
    inactives: ClassVar[QuerySetManager] = InactiveManager()

Agora vamos criar um novo gestor e usá-lo com nosso exemplo anterior.

from typing import ClassVar

import mongoz
from mongoz import QuerySetManager

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


class InactiveManager(QuerySetManager):
    """
    Custom manager that will return only active users
    """

    def get_queryset(self) -> "QuerySetManager":
        queryset = super().get_queryset().filter(is_active=False)
        return queryset


class User(mongoz.Document):
    # Add the new manager
    inactives: ClassVar[QuerySetManager] = InactiveManager()

    name: str = mongoz.String(max_length=255)
    email: str = mongoz.Email(max_length=70)
    is_active: bool = mongoz.Boolean(default=True)


# Create an inactive user
await User.objects.create(name="Edgy", email="foo@bar.com", is_active=False)  # noqa

# You can also create a user using the new manager
await User.inactives.create(name="Another Edgy", email="bar@foo.com", is_active=False)  # noqa

# Querying using the new manager
user = await User.inactives.get(email="foo@bar.com")  # noqa
# User(ObjectId(...))

user = await User.inactives.get(email="bar@foo.com")  # noqa
# User(ObjectId(...))

# Create a user using the default manager
await User.objects.create(name="Edgy", email="user@edgy.com")  # noqa

# Querying all inactives only
users = await User.inactives.all()  # noqa
# [User(ObjectId(...)), User(ObjectId(...))]

# Querying them all
user = await User.objects.all()  # noqa
# [User(ObjectId(...)), User(ObjectId(...)), User(ObjectId(...))]

Estes gestores podem ser tão complexos quanto desejar, com quantos filtros desejar. O que precisa fazer é simplesmente substituir o get_queryset() e adicioná-lo aos seus documentos.

Substituir o gestor padrão

Também é possível substituir o gestor padrão criando o gestor personalizado e substituindo o gestor objects.

from typing import ClassVar

import mongoz
from mongoz import QuerySetManager

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


class InactiveManager(QuerySetManager):
    """
    Custom manager that will return only active users
    """

    def get_queryset(self) -> "QuerySetManager":
        queryset = super().get_queryset().filter(is_active=False)
        return queryset


class User(mongoz.Document):
    # Add the new manager
    objects: ClassVar[QuerySetManager] = InactiveManager()

    name: str = mongoz.String(max_length=255)
    email: str = mongoz.Email(max_length=70)
    is_active: bool = mongoz.Boolean(default=True)


# Create an inactive user
await User.objects.create(name="Edgy", email="foo@bar.com", is_active=False)  # noqa

# You can also create a user using the new manager
await User.objects.create(name="Another Edgy", email="bar@foo.com", is_active=False)  # noqa

# Create a user using the default manager
await User.objects.create(name="Edgy", email="user@edgy.com")  # noqa

# Querying them all
user = await User.objects.all()  # noqa
# [User(ObjectId(...)), User(ObjectId(...))]

Warning

Tenha cuidado ao substituir o gestor padrão, pois pode não obter todos os resultados do .all() se não filtrar correctamente.