API Reference¶

Intro¶

In this tutorial section, we will learn everything in great detail.

Sometimes we need to check a function or variable as a reference.

In this section, we will try to create a cheat sheet for quick reference.

Insert Some Data¶

import os
from typing import Optional

from mongodb_odm import ASCENDING, Document, IndexModel, apply_indexes, connect


class Player(Document):
    name: str
    country_code: str
    rating: Optional[int] = None

    class ODMConfig(Document.ODMConfig):
        indexes = [
            IndexModel([("rating", ASCENDING)]),
        ]


def create_players():
    Player(name="Pelé", country_code="BRA", rating=98).create()
    Player(name="Diego Maradona", country_code="ARG", rating=97).create()
    Player(name="Zinedine Zidane", country_code="FRA", rating=94).create()
    Player(name="Ronaldo", country_code="BRA", rating=94).create()
    Player(name="Neymar", country_code="BRA", rating=89).create()
    Player(name="Lionel Messi", country_code="ARG", rating=91).create()
    Player(name="Ángel Di María", country_code="ARG", rating=84).create()
    Player(name="Karim Benzema", country_code="FRA", rating=89).create()
    Player(name="Antoine Griezmann", country_code="FRA", rating=85).create()
    Player(name="Kylian Mbappé", country_code="FRA", rating=91).create()
    Player(name="Gerd Müller", country_code="GER").create()
    Player(name="Miroslav Klose", country_code="GER", rating=91).create()
    Player(name="Thomas Müller", country_code="GER", rating=87).create()
    Player(name="Cristiano Ronaldo", country_code="POR", rating=87).create()
    Player(name="Eusébio", country_code="POR", rating=93).create()
    Player(name="Diogo Jota", country_code="POR", rating=85).create()
    Player(name="David Beckham", country_code="ENG", rating=89).create()
    Player(name="Wayne Rooney", country_code="ENG", rating=80).create()
    Player(name="Harry Kane", country_code="ENG", rating=89).create()


def main():
    connect(os.environ.get("MONGO_URL", "mongodb://localhost:27017/testdb"))
    apply_indexes()
    create_players()


if __name__ == "__main__":
    main()

Connection¶

connect¶

def connect(url: str, databases: Optional[Set[str]] = None) -> MongoClient[Any]:

disconnect¶

def disconnect() -> bool:

get_client¶

def get_client() -> MongoClient[Any]:

Definition of Model Class¶

Class¶

ODMConfig¶

In the ODMConfig class, we already set some default values to change class behavior.

    class ODMConfig(BaseModel.ODMConfig):
        collection_name: Optional[str] = None
        allow_inheritance: bool = False
        index_inheritance_field: bool = True

collection_name type Optional[str] default None
allow_inheritance type bool default False
index_inheritance_field type bool default True

Types¶

Generic dict Type¶

We will use this DICT_TYPE for all of our generic dict types.

DICT_TYPE = dict[str, Any]

Sort Type¶

We will use this SORT_TYPE in all of our sorting arguments.

SORT_TYPE = Union[str, Sequence[tuple[str, Union[int, str, Mapping[str, Any]]]]]

Class Methods¶

In this section, we will explain every Class Method that is callable from a defined class.

_db¶

@classmethod
def _db(cls) -> str:

We can get the collection name by calling _db.

create¶

def create(self, **kwargs) -> Self:

The create method does not accept any MongoDB-ODM related arguments. Its kwargs are directly sent to the PyMongo function.

find_raw¶

@classmethod
def find_raw(
    cls,
    filter: Optional[DICT_TYPE] = None,
    projection: Optional[DICT_TYPE] = None,
    **kwargs: Any,
) -> Cursor[Any]:

The find_raw is a special function that returns a PyMongo find cursor.

find¶

@classmethod
def find(
    cls,
    filter: Optional[DICT_TYPE] = None,
    projection: Optional[DICT_TYPE] = None,
    sort: Optional[SORT_TYPE] = None,
    skip: Optional[int] = None,
    limit: Optional[int] = None,
    **kwargs: Any,
) -> Iterator[Self]:

Parameters¶

The find classmethod accepts 6 parameters.

filter - The data type should be dict and the default value is None.
projection - The data type should be dict and the default value is None.
sort - The data type should be Optional[SORT_TYPE] and the default value is None.
skip - The data type is Optional[int] and the default value is None.
limit - The data type is Optional[int] and the default value is None.
Lastly, it accepts **kwargs

Return Type¶

It returns Iterator[Self], which is iterable. You should get type support on each object.

find_one¶

@classmethod
def find_one(
    cls,
    filter: Optional[DICT_TYPE] = None,
    projection: Optional[DICT_TYPE] = None,
    sort: Optional[SORT_TYPE] = None,
    **kwargs: Any,
) -> Optional[Self]:

find_one Parameters¶

The find_one classmethod accepts 4 parameters.

filter - The data type should be dict and the default value is None.
projection - The data type should be dict and the default value is None.
sort - The data type should be Optional[SORT_TYPE] and the default value is None.
Lastly, it accepts **kwargs

get¶

@classmethod
def get(
    cls,
    filter: DICT_TYPE,
    sort: Optional[SORT_TYPE] = None,
    **kwargs: Any,
) -> Self:

get_or_create¶

@classmethod
def get_or_create(
    cls,
    filter: DICT_TYPE,
    sort: Optional[SORT_TYPE] = None,
    **kwargs: Any,
) -> tuple[Self, bool]:

count_documents¶

@classmethod
def count_documents(cls, filter: Optional[DICT_TYPE] = None, **kwargs: Any) -> int:

exists¶

@classmethod
def exists(cls, filter: Optional[DICT_TYPE] = None, **kwargs: Any) -> bool:

aggregate¶

@classmethod
def aggregate(
    cls,
    pipeline: list[Any],
    get_raw: bool = False,
    inheritance_filter: bool = True,
    **kwargs: Any,
) -> Iterator[Any]:

get_random_one¶

@classmethod
def get_random_one(cls, filter: Optional[DICT_TYPE] = None, **kwargs: Any) -> Self:

update_one¶

@classmethod
def update_one(
    cls, filter: DICT_TYPE, data: DICT_TYPE, **kwargs: Any
) -> UpdateResult:

update_many¶

@classmethod
def update_many(
    cls, filter: DICT_TYPE, data: DICT_TYPE, **kwargs: Any
) -> UpdateResult:

delete_one¶

@classmethod
def delete_one(cls, filter: DICT_TYPE, **kwargs: Any) -> DeleteResult:

delete_many¶

@classmethod
def delete_many(cls, filter: DICT_TYPE, **kwargs: Any) -> DeleteResult:

bulk_write¶

@classmethod
def bulk_write(
    cls, requests: Sequence[WriteOp[Any]], **kwargs: Any
) -> BulkWriteResult:

Here WriteOp is one of the types: DeleteMany, DeleteOne, IndexModel, InsertOne, ReplaceOne, UpdateMany, UpdateOne from pymongo

load_related¶

The classmethod load_related will load all related fields from the database.

Data will be loaded immediately; no lazy loading will be applied.

@classmethod
def load_related(
    cls,
    object_list: Union[Iterator[Self], Sequence[Self]],
    fields: Optional[list[str]] = None,
    **kwargs: Any,
) -> Sequence[Self]:

Methods¶

In this section we will explain every Method that is callable after creating the object.

update¶

def update(self, raw: Optional[DICT_TYPE] = None, **kwargs: Any) -> UpdateResult:

delete¶

def delete(self, **kwargs: Any) -> DeleteResult: