Foreign Key Relationship Intro

Foreign Key Relation

We added a new Country model that has a relationship with the Player model.

To create a relationship with the Player model, we need to use ODMObjectId as the data type, which is imported from mongodb_odm.

import os
from typing import Optional

from mongodb_odm import (
    ASCENDING,
    Document,
    IndexModel,
    ODMObjectId,
    Relationship,
    apply_indexes,
    connect,
)


class Country(Document):
    name: str


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

    country: Optional[Country] = Relationship(local_field="country_id")

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

# Code omitted below

Full file preview

import os
from typing import Optional

from mongodb_odm import (
    ASCENDING,
    Document,
    IndexModel,
    ODMObjectId,
    Relationship,
    apply_indexes,
    connect,
)


class Country(Document):
    name: str


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

    country: Optional[Country] = Relationship(local_field="country_id")

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


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


def insert_data():
    bangladesh = Country(name="Bangladesh").create()

    Player(name="Jamal Bhuyan", country_id=bangladesh.id).create()
    Player(name="Mohamed Emon Mahmud", country_id=bangladesh.id).create()


def read_data():
    for player in Player.find():
        print(player)
    print()


def read_data_with_related_field():
    player_qs = Player.find()
    for player in Player.load_related(player_qs):
        print(player)
    print()


def read_related_data_of_specific_field():
    player_qs = Player.find()
    for player in Player.load_related(player_qs, fields=["country"]):
        print(player)
    print()


def read_related_data_for_single_obj():
    """Load related data in naive way"""
    player = Player.find_one()
    if player:
        player.country = Country.find_one({"_id": player.country_id})
    print(player)

    """Load related data with load_related"""
    player = Player.find_one()
    if player:
        player = Player.load_related([player])[0]
    print(player)


def main():
    configuration()
    insert_data()

    read_data()
    read_data_with_related_field()
    read_related_data_of_specific_field()
    read_related_data_for_single_obj()


if __name__ == "__main__":
    main()

Tip

MongoDB does not manage or validate relationships between collections. We have to manage that data ourselves. We are assigning a foreign key with ODMObjectId.

First import ODMObjectId from mongodb_odm.

ODMObjectId has the complete functionality of ObjectId from the bson package. ODMObjectId was directly inherited from ObjectId and added a validation function to work with Pydantic.

Relationship

From the Player model we declare country_id which is an ODMObjectId type object. The country_id field only holds the _id from the country document.

Here we define a logical field country. The country field doesn't have any action in the database.

The Relationship accepts multiple fields, one of them is local_field and it's required. The local_field field will define the local field that is related to.

import os
from typing import Optional

from mongodb_odm import (
    ASCENDING,
    Document,
    IndexModel,
    ODMObjectId,
    Relationship,
    apply_indexes,
    connect,
)


class Country(Document):
    name: str


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

    country: Optional[Country] = Relationship(local_field="country_id")

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

# Code omitted below

Full file preview

import os
from typing import Optional

from mongodb_odm import (
    ASCENDING,
    Document,
    IndexModel,
    ODMObjectId,
    Relationship,
    apply_indexes,
    connect,
)


class Country(Document):
    name: str


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

    country: Optional[Country] = Relationship(local_field="country_id")

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


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


def insert_data():
    bangladesh = Country(name="Bangladesh").create()

    Player(name="Jamal Bhuyan", country_id=bangladesh.id).create()
    Player(name="Mohamed Emon Mahmud", country_id=bangladesh.id).create()


def read_data():
    for player in Player.find():
        print(player)
    print()


def read_data_with_related_field():
    player_qs = Player.find()
    for player in Player.load_related(player_qs):
        print(player)
    print()


def read_related_data_of_specific_field():
    player_qs = Player.find()
    for player in Player.load_related(player_qs, fields=["country"]):
        print(player)
    print()


def read_related_data_for_single_obj():
    """Load related data in naive way"""
    player = Player.find_one()
    if player:
        player.country = Country.find_one({"_id": player.country_id})
    print(player)

    """Load related data with load_related"""
    player = Player.find_one()
    if player:
        player = Player.load_related([player])[0]
    print(player)


def main():
    configuration()
    insert_data()

    read_data()
    read_data_with_related_field()
    read_related_data_of_specific_field()
    read_related_data_for_single_obj()


if __name__ == "__main__":
    main()

Insert Player

In this example, we will create some players that have a relationship with the country collection.

# Code omitted above

def insert_data():
    bangladesh = Country(name="Bangladesh").create()

    Player(name="Jamal Bhuyan", country_id=bangladesh.id).create()
    Player(name="Mohamed Emon Mahmud", country_id=bangladesh.id).create()

# Code omitted below

Full file preview

import os
from typing import Optional

from mongodb_odm import (
    ASCENDING,
    Document,
    IndexModel,
    ODMObjectId,
    Relationship,
    apply_indexes,
    connect,
)


class Country(Document):
    name: str


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

    country: Optional[Country] = Relationship(local_field="country_id")

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


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


def insert_data():
    bangladesh = Country(name="Bangladesh").create()

    Player(name="Jamal Bhuyan", country_id=bangladesh.id).create()
    Player(name="Mohamed Emon Mahmud", country_id=bangladesh.id).create()


def read_data():
    for player in Player.find():
        print(player)
    print()


def read_data_with_related_field():
    player_qs = Player.find()
    for player in Player.load_related(player_qs):
        print(player)
    print()


def read_related_data_of_specific_field():
    player_qs = Player.find()
    for player in Player.load_related(player_qs, fields=["country"]):
        print(player)
    print()


def read_related_data_for_single_obj():
    """Load related data in naive way"""
    player = Player.find_one()
    if player:
        player.country = Country.find_one({"_id": player.country_id})
    print(player)

    """Load related data with load_related"""
    player = Player.find_one()
    if player:
        player = Player.load_related([player])[0]
    print(player)


def main():
    configuration()
    insert_data()

    read_data()
    read_data_with_related_field()
    read_related_data_of_specific_field()
    read_related_data_for_single_obj()


if __name__ == "__main__":
    main()

First we clear all data from the Country and Player collections.

Then we create a document for Country.

After that, we create two players in the database that have a relationship with the country document.

Read Data

Here we read all players from the database. But the country field will not have a Country object since we don't read it from the database.

# Code omitted above

def read_data():
    for player in Player.find():
        print(player)
    print()

# Code omitted below

Full file preview

import os
from typing import Optional

from mongodb_odm import (
    ASCENDING,
    Document,
    IndexModel,
    ODMObjectId,
    Relationship,
    apply_indexes,
    connect,
)


class Country(Document):
    name: str


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

    country: Optional[Country] = Relationship(local_field="country_id")

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


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


def insert_data():
    bangladesh = Country(name="Bangladesh").create()

    Player(name="Jamal Bhuyan", country_id=bangladesh.id).create()
    Player(name="Mohamed Emon Mahmud", country_id=bangladesh.id).create()


def read_data():
    for player in Player.find():
        print(player)
    print()


def read_data_with_related_field():
    player_qs = Player.find()
    for player in Player.load_related(player_qs):
        print(player)
    print()


def read_related_data_of_specific_field():
    player_qs = Player.find()
    for player in Player.load_related(player_qs, fields=["country"]):
        print(player)
    print()


def read_related_data_for_single_obj():
    """Load related data in naive way"""
    player = Player.find_one()
    if player:
        player.country = Country.find_one({"_id": player.country_id})
    print(player)

    """Load related data with load_related"""
    player = Player.find_one()
    if player:
        player = Player.load_related([player])[0]
    print(player)


def main():
    configuration()
    insert_data()

    read_data()
    read_data_with_related_field()
    read_related_data_of_specific_field()
    read_related_data_for_single_obj()


if __name__ == "__main__":
    main()

Load Related Field

Read data with related field

In this example, we will read all players with their related countries.

To read related countries we will use an extra class method load_related.

The load_related is a classmethod that loads all or partially related fields and allocates them with related fields.

# Code omitted above

def read_data_with_related_field():
    player_qs = Player.find()
    for player in Player.load_related(player_qs):
        print(player)
    print()

# Code omitted below

Full file preview

import os
from typing import Optional

from mongodb_odm import (
    ASCENDING,
    Document,
    IndexModel,
    ODMObjectId,
    Relationship,
    apply_indexes,
    connect,
)


class Country(Document):
    name: str


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

    country: Optional[Country] = Relationship(local_field="country_id")

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


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


def insert_data():
    bangladesh = Country(name="Bangladesh").create()

    Player(name="Jamal Bhuyan", country_id=bangladesh.id).create()
    Player(name="Mohamed Emon Mahmud", country_id=bangladesh.id).create()


def read_data():
    for player in Player.find():
        print(player)
    print()


def read_data_with_related_field():
    player_qs = Player.find()
    for player in Player.load_related(player_qs):
        print(player)
    print()


def read_related_data_of_specific_field():
    player_qs = Player.find()
    for player in Player.load_related(player_qs, fields=["country"]):
        print(player)
    print()


def read_related_data_for_single_obj():
    """Load related data in naive way"""
    player = Player.find_one()
    if player:
        player.country = Country.find_one({"_id": player.country_id})
    print(player)

    """Load related data with load_related"""
    player = Player.find_one()
    if player:
        player = Player.load_related([player])[0]
    print(player)


def main():
    configuration()
    insert_data()

    read_data()
    read_data_with_related_field()
    read_related_data_of_specific_field()
    read_related_data_for_single_obj()


if __name__ == "__main__":
    main()

After running the read_data_with_related_field function two objects will be printed with related data country.

Player(id=ObjectId('id'), name='Jamal Bhuyan', country_id=ObjectId('id'), rating=None, country=Country(id=ObjectId('id'), name='Bangladesh', _id=ObjectId('id')), _id=ObjectId('id'))
Player(id=ObjectId('id'), name='Mohamed Emon Mahmud', country_id=ObjectId('id'), rating=None, country=Country(id=ObjectId('id'), name='Bangladesh', _id=ObjectId('id')), _id=ObjectId('id'))

The printed data is displayed again with pretty formatting.

Player(
    id=ObjectId("id"),
    name="Jamal Bhuyan",
    country_id=ObjectId("id"),
    rating=None,
    country=Country(
        id=ObjectId("id"),
        name="Bangladesh",
        _id=ObjectId("id"),
    ),
    _id=ObjectId("id"),
)
Player(
    id=ObjectId("id"),
    name="Mohamed Emon Mahmud",
    country_id=ObjectId("id"),
    rating=None,
    country=Country(
        id=ObjectId("id"),
        name="Bangladesh",
        _id=ObjectId("id"),
    ),
    _id=ObjectId("id"),
)

Warning

If we call load_related, the data will be loaded immediately; no lazy loading will be applied.

Read related data of specific field

By default, load_related will load all related data from the database.

We can narrow down which relational documents will be loaded from the database by passing a list of field names in fields if we don't need all of them.

In this example, we pass the country field that needs to be loaded from the database.

# Code omitted above

def read_related_data_of_specific_field():
    player_qs = Player.find()
    for player in Player.load_related(player_qs, fields=["country"]):
        print(player)
    print()

# Code omitted below

Full file preview

import os
from typing import Optional

from mongodb_odm import (
    ASCENDING,
    Document,
    IndexModel,
    ODMObjectId,
    Relationship,
    apply_indexes,
    connect,
)


class Country(Document):
    name: str


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

    country: Optional[Country] = Relationship(local_field="country_id")

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


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


def insert_data():
    bangladesh = Country(name="Bangladesh").create()

    Player(name="Jamal Bhuyan", country_id=bangladesh.id).create()
    Player(name="Mohamed Emon Mahmud", country_id=bangladesh.id).create()


def read_data():
    for player in Player.find():
        print(player)
    print()


def read_data_with_related_field():
    player_qs = Player.find()
    for player in Player.load_related(player_qs):
        print(player)
    print()


def read_related_data_of_specific_field():
    player_qs = Player.find()
    for player in Player.load_related(player_qs, fields=["country"]):
        print(player)
    print()


def read_related_data_for_single_obj():
    """Load related data in naive way"""
    player = Player.find_one()
    if player:
        player.country = Country.find_one({"_id": player.country_id})
    print(player)

    """Load related data with load_related"""
    player = Player.find_one()
    if player:
        player = Player.load_related([player])[0]
    print(player)


def main():
    configuration()
    insert_data()

    read_data()
    read_data_with_related_field()
    read_related_data_of_specific_field()
    read_related_data_for_single_obj()


if __name__ == "__main__":
    main()

After running the read_related_data_of_specific_field function the output will be the same as the previous function since we don't have multiple related fields.

Load Related for Single Object

We can also load related data for single objects.

In this example, we provide two approaches to load related data.

In approach one, we use the general method.

In approach two, we use the classmethod load_related to load related data.

# Code omitted above

def read_related_data_for_single_obj():
    """Load related data in naive way"""
    player = Player.find_one()
    if player:
        player.country = Country.find_one({"_id": player.country_id})
    print(player)

    """Load related data with load_related"""
    player = Player.find_one()
    if player:
        player = Player.load_related([player])[0]
    print(player)

# Code omitted below

Full file preview

import os
from typing import Optional

from mongodb_odm import (
    ASCENDING,
    Document,
    IndexModel,
    ODMObjectId,
    Relationship,
    apply_indexes,
    connect,
)


class Country(Document):
    name: str


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

    country: Optional[Country] = Relationship(local_field="country_id")

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


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


def insert_data():
    bangladesh = Country(name="Bangladesh").create()

    Player(name="Jamal Bhuyan", country_id=bangladesh.id).create()
    Player(name="Mohamed Emon Mahmud", country_id=bangladesh.id).create()


def read_data():
    for player in Player.find():
        print(player)
    print()


def read_data_with_related_field():
    player_qs = Player.find()
    for player in Player.load_related(player_qs):
        print(player)
    print()


def read_related_data_of_specific_field():
    player_qs = Player.find()
    for player in Player.load_related(player_qs, fields=["country"]):
        print(player)
    print()


def read_related_data_for_single_obj():
    """Load related data in naive way"""
    player = Player.find_one()
    if player:
        player.country = Country.find_one({"_id": player.country_id})
    print(player)

    """Load related data with load_related"""
    player = Player.find_one()
    if player:
        player = Player.load_related([player])[0]
    print(player)


def main():
    configuration()
    insert_data()

    read_data()
    read_data_with_related_field()
    read_related_data_of_specific_field()
    read_related_data_for_single_obj()


if __name__ == "__main__":
    main()

The printed data from the two methods will be identical.

Also, the two methods are functionally almost identical.