Managers

The managers are a great tool that Mongoz offers. Heavily inspired by Edgy, the managers allow you to build unique tailored queries ready to be used by your documents.

Mongoz by default uses the the manager called objects which it makes it simple to understand.

Let us see an example.

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

When querying the User table, the objects (manager) is the default and should be always presented when doing it so.

Danger

This is only applied to the manager side of Mongoz, for example, when using Document.objects..... When using Document.query... it will not work.

Custom manager

It is also possible to have your own custom managers and to do it so, you should inherit the QuerySetManager class and override the get_queryset().

For those familiar with Django managers, the principle is exactly the same. 😀

The managers must be type annotated ClassVar or an error be raised.

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

Let us now create new manager and use it with our previous example.

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

These managers can be as complex as you like with as many filters as you desire. What you need is simply override the get_queryset() and add it to your documents.

Override the default manager

Overriding the default manager is also possible by creating the custom manager and overriding the objects manager.

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

Be careful when overriding the default manager as you might not get all the results from the .all() if you don't filter properly.