Many to Many Relationship Intro

In a traditional SQL database, we use a 3rd table to implement many-to-many relations, where the middle table holds the id from table A and the id from table B.

In this chapter, we will add skills for players.

In MongoDB, we can implement many-to-many relations using two-way embedded approaches and traditional approaches. MongoDB won't block us from using any one of these.

We add another model Skill alongside Country and Player to illustrate many-to-many relations.

Model definition

First, we will implement the traditional way of implementing many-to-many relations.

First, declare the Country, Skill, and Player models.

Then, declare the middle model PlayerSkill that will have player_id, skill_id, and rating fields.

# Code omitted above

class Country(Document):
    name: str


class Skill(Document):
    name: str


class Player(Document):
    name: str
    country_id: ODMObjectId

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

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


class PlayerSkill(Document):
    player_id: ODMObjectId
    skill_id: ODMObjectId
    rating: int = Field(default=0, ge=1, le=100)

    player: Optional[Player] = Relationship(local_field="player_id")
    skill: Optional[Skill] = Relationship(local_field="skill_id")

    class ODMConfig(Document.ODMConfig):
        indexes = [
            IndexModel(
                [("player_id", ASCENDING), ("skill_id", ASCENDING)],
                unique=True,
            ),
        ]

# Code omitted below

Full file preview

import os
from typing import Optional

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


class Country(Document):
    name: str


class Skill(Document):
    name: str


class Player(Document):
    name: str
    country_id: ODMObjectId

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

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


class PlayerSkill(Document):
    player_id: ODMObjectId
    skill_id: ODMObjectId
    rating: int = Field(default=0, ge=1, le=100)

    player: Optional[Player] = Relationship(local_field="player_id")
    skill: Optional[Skill] = Relationship(local_field="skill_id")

    class ODMConfig(Document.ODMConfig):
        indexes = [
            IndexModel(
                [("player_id", ASCENDING), ("skill_id", ASCENDING)],
                unique=True,
            ),
        ]


def insert_data():
    brazil = Country(name="Brazil").create()
    argentina = Country(name="Argentina").create()
    france = Country(name="France").create()

    catching = Skill(name="Catching")
    running = Skill(name="Running")
    kicking = Skill(name="Kicking")

    pele = Player(name="Pelé", country_id=brazil.id).create()
    PlayerSkill(player_id=pele.id, skill_id=catching.id, rating=49).create()
    PlayerSkill(player_id=pele.id, skill_id=kicking.id, rating=49).create()

    maradona = Player(name="Diego Maradona", country_id=argentina.id).create()
    PlayerSkill(player_id=maradona.id, skill_id=catching.id, rating=48).create()
    PlayerSkill(player_id=maradona.id, skill_id=kicking.id, rating=49).create()

    zidane = Player(name="Zinedine Zidane", country_id=france.id).create()
    PlayerSkill(player_id=zidane.id, skill_id=running.id, rating=42).create()
    PlayerSkill(player_id=zidane.id, skill_id=kicking.id, rating=42).create()


def read_data():
    pele = Player.get({Player.name: "Pelé"})
    skills = list(PlayerSkill.find({"player_id": pele.id}))

    print(pele)
    print(skills)


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


def main():
    configuration()
    insert_data()
    read_data()


if __name__ == "__main__":
    main()

Multi key indexes

We will define a multi-key index for better performance. Also, we want player_id and skill_id to be unique.

# Code omitted above

class PlayerSkill(Document):
    player_id: ODMObjectId
    skill_id: ODMObjectId
    rating: int = Field(default=0, ge=1, le=100)

    player: Optional[Player] = Relationship(local_field="player_id")
    skill: Optional[Skill] = Relationship(local_field="skill_id")

    class ODMConfig(Document.ODMConfig):
        indexes = [
            IndexModel(
                [("player_id", ASCENDING), ("skill_id", ASCENDING)],
                unique=True,
            ),
        ]

# Code omitted below

Here we use IndexModel, which is directly used from Pymongo.

Read data

We will get one of the player's data and his skills.

# Code omitted above


def read_data():
    pele = Player.get({Player.name: "Pelé"})
    skills = list(PlayerSkill.find({"player_id": pele.id}))

    print(pele)
    print(skills)

# Code omitted below

Full file preview

import os
from typing import Optional

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


class Country(Document):
    name: str


class Skill(Document):
    name: str


class Player(Document):
    name: str
    country_id: ODMObjectId

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

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


class PlayerSkill(Document):
    player_id: ODMObjectId
    skill_id: ODMObjectId
    rating: int = Field(default=0, ge=1, le=100)

    player: Optional[Player] = Relationship(local_field="player_id")
    skill: Optional[Skill] = Relationship(local_field="skill_id")

    class ODMConfig(Document.ODMConfig):
        indexes = [
            IndexModel(
                [("player_id", ASCENDING), ("skill_id", ASCENDING)],
                unique=True,
            ),
        ]


def insert_data():
    brazil = Country(name="Brazil").create()
    argentina = Country(name="Argentina").create()
    france = Country(name="France").create()

    catching = Skill(name="Catching")
    running = Skill(name="Running")
    kicking = Skill(name="Kicking")

    pele = Player(name="Pelé", country_id=brazil.id).create()
    PlayerSkill(player_id=pele.id, skill_id=catching.id, rating=49).create()
    PlayerSkill(player_id=pele.id, skill_id=kicking.id, rating=49).create()

    maradona = Player(name="Diego Maradona", country_id=argentina.id).create()
    PlayerSkill(player_id=maradona.id, skill_id=catching.id, rating=48).create()
    PlayerSkill(player_id=maradona.id, skill_id=kicking.id, rating=49).create()

    zidane = Player(name="Zinedine Zidane", country_id=france.id).create()
    PlayerSkill(player_id=zidane.id, skill_id=running.id, rating=42).create()
    PlayerSkill(player_id=zidane.id, skill_id=kicking.id, rating=42).create()


def read_data():
    pele = Player.get({Player.name: "Pelé"})
    skills = list(PlayerSkill.find({"player_id": pele.id}))

    print(pele)
    print(skills)


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


def main():
    configuration()
    insert_data()
    read_data()


if __name__ == "__main__":
    main()

After running the read_data function, the console should print:

Player(id=ObjectId('id'), name='Pelé', country_id=ObjectId('id'), _id=ObjectId('id'))

[PlayerSkill(id=ObjectId('id'), player_id=ObjectId('id'), skill_id=ObjectId('id'), rating=49, _id=ObjectId('id')), PlayerSkill(id=ObjectId('id'), player_id=ObjectId('id'), skill_id=ObjectId('id'), rating=49, _id=ObjectId('id'))]

Run Full Code

Run the code and check the MongoDB document viewer to see the impact.

import os
from typing import Optional

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


class Country(Document):
    name: str


class Skill(Document):
    name: str


class Player(Document):
    name: str
    country_id: ODMObjectId

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

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


class PlayerSkill(Document):
    player_id: ODMObjectId
    skill_id: ODMObjectId
    rating: int = Field(default=0, ge=1, le=100)

    player: Optional[Player] = Relationship(local_field="player_id")
    skill: Optional[Skill] = Relationship(local_field="skill_id")

    class ODMConfig(Document.ODMConfig):
        indexes = [
            IndexModel(
                [("player_id", ASCENDING), ("skill_id", ASCENDING)],
                unique=True,
            ),
        ]


def insert_data():
    brazil = Country(name="Brazil").create()
    argentina = Country(name="Argentina").create()
    france = Country(name="France").create()

    catching = Skill(name="Catching")
    running = Skill(name="Running")
    kicking = Skill(name="Kicking")

    pele = Player(name="Pelé", country_id=brazil.id).create()
    PlayerSkill(player_id=pele.id, skill_id=catching.id, rating=49).create()
    PlayerSkill(player_id=pele.id, skill_id=kicking.id, rating=49).create()

    maradona = Player(name="Diego Maradona", country_id=argentina.id).create()
    PlayerSkill(player_id=maradona.id, skill_id=catching.id, rating=48).create()
    PlayerSkill(player_id=maradona.id, skill_id=kicking.id, rating=49).create()

    zidane = Player(name="Zinedine Zidane", country_id=france.id).create()
    PlayerSkill(player_id=zidane.id, skill_id=running.id, rating=42).create()
    PlayerSkill(player_id=zidane.id, skill_id=kicking.id, rating=42).create()


def read_data():
    pele = Player.get({Player.name: "Pelé"})
    skills = list(PlayerSkill.find({"player_id": pele.id}))

    print(pele)
    print(skills)


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


def main():
    configuration()
    insert_data()
    read_data()


if __name__ == "__main__":
    main()

Model definition (Embedded way)

In this example, we use MongoDB's embedded approach to implement many-to-many relations.

# Code omitted above

class Country(Document):
    name: str


class Skill(Document):
    name: str


class EmbeddedSkill(BaseModel):
    skill_id: ODMObjectId
    rating: int = Field(default=0, ge=1, le=100)


class Player(Document):
    name: str
    country_id: ODMObjectId
    skills: list[EmbeddedSkill] = []

    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,
    BaseModel,
    Document,
    Field,
    IndexModel,
    ODMObjectId,
    Relationship,
    apply_indexes,
    connect,
)


class Country(Document):
    name: str


class Skill(Document):
    name: str


class EmbeddedSkill(BaseModel):
    skill_id: ODMObjectId
    rating: int = Field(default=0, ge=1, le=100)


class Player(Document):
    name: str
    country_id: ODMObjectId
    skills: list[EmbeddedSkill] = []

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

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


def insert_data():
    brazil = Country(name="Brazil").create()
    argentina = Country(name="Argentina").create()
    france = Country(name="France").create()

    catching = Skill(name="Catching")
    running = Skill(name="Running")
    kicking = Skill(name="Kicking")

    Player(
        name="Pelé",
        country_id=brazil.id,
        skills=[
            EmbeddedSkill(skill_id=catching.id, rating=49),
            EmbeddedSkill(skill_id=kicking.id, rating=49),
        ],
    ).create()
    Player(
        name="Diego Maradona",
        country_id=argentina.id,
        skills=[
            EmbeddedSkill(skill_id=catching.id, rating=48),
            EmbeddedSkill(skill_id=kicking.id, rating=49),
        ],
    ).create()
    Player(
        name="Zinedine Zidane",
        country_id=france.id,
        skills=[
            EmbeddedSkill(skill_id=running.id, rating=42),
            EmbeddedSkill(skill_id=kicking.id, rating=42),
        ],
    ).create()


def read_data():
    pele = Player.get({Player.name: "Pelé"})
    print(pele)


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


def main():
    configuration()
    insert_data()
    read_data()


if __name__ == "__main__":
    main()

First, we define the new model Skill.

Then we define the Pydantic model EmbeddedSkill. The name is not special and we can use any name we want.

The EmbeddedSkill has two fields: ODMObjectId type skill_id and int type rating.

We add the list[EmbeddedSkill] type skills field in the Player model.

Insert Data

First, create a country and skill document.

Then create a player with a list of EmbeddedSkill.

# Code omitted above

def insert_data():
    brazil = Country(name="Brazil").create()
    argentina = Country(name="Argentina").create()
    france = Country(name="France").create()

    catching = Skill(name="Catching")
    running = Skill(name="Running")
    kicking = Skill(name="Kicking")

    Player(
        name="Pelé",
        country_id=brazil.id,
        skills=[
            EmbeddedSkill(skill_id=catching.id, rating=49),
            EmbeddedSkill(skill_id=kicking.id, rating=49),
        ],
    ).create()
    Player(
        name="Diego Maradona",
        country_id=argentina.id,
        skills=[
            EmbeddedSkill(skill_id=catching.id, rating=48),
            EmbeddedSkill(skill_id=kicking.id, rating=49),
        ],
    ).create()
    Player(
        name="Zinedine Zidane",
        country_id=france.id,
        skills=[
            EmbeddedSkill(skill_id=running.id, rating=42),
            EmbeddedSkill(skill_id=kicking.id, rating=42),
        ],
    ).create()

# Code omitted below

Full file preview

import os
from typing import Optional

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


class Country(Document):
    name: str


class Skill(Document):
    name: str


class EmbeddedSkill(BaseModel):
    skill_id: ODMObjectId
    rating: int = Field(default=0, ge=1, le=100)


class Player(Document):
    name: str
    country_id: ODMObjectId
    skills: list[EmbeddedSkill] = []

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

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


def insert_data():
    brazil = Country(name="Brazil").create()
    argentina = Country(name="Argentina").create()
    france = Country(name="France").create()

    catching = Skill(name="Catching")
    running = Skill(name="Running")
    kicking = Skill(name="Kicking")

    Player(
        name="Pelé",
        country_id=brazil.id,
        skills=[
            EmbeddedSkill(skill_id=catching.id, rating=49),
            EmbeddedSkill(skill_id=kicking.id, rating=49),
        ],
    ).create()
    Player(
        name="Diego Maradona",
        country_id=argentina.id,
        skills=[
            EmbeddedSkill(skill_id=catching.id, rating=48),
            EmbeddedSkill(skill_id=kicking.id, rating=49),
        ],
    ).create()
    Player(
        name="Zinedine Zidane",
        country_id=france.id,
        skills=[
            EmbeddedSkill(skill_id=running.id, rating=42),
            EmbeddedSkill(skill_id=kicking.id, rating=42),
        ],
    ).create()


def read_data():
    pele = Player.get({Player.name: "Pelé"})
    print(pele)


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


def main():
    configuration()
    insert_data()
    read_data()


if __name__ == "__main__":
    main()

Read Data

Let's get one of the player's data.

# Code omitted above

def read_data():
    pele = Player.get({Player.name: "Pelé"})
    print(pele)

# Code omitted below

Full file preview

import os
from typing import Optional

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


class Country(Document):
    name: str


class Skill(Document):
    name: str


class Player(Document):
    name: str
    country_id: ODMObjectId

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

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


class PlayerSkill(Document):
    player_id: ODMObjectId
    skill_id: ODMObjectId
    rating: int = Field(default=0, ge=1, le=100)

    player: Optional[Player] = Relationship(local_field="player_id")
    skill: Optional[Skill] = Relationship(local_field="skill_id")

    class ODMConfig(Document.ODMConfig):
        indexes = [
            IndexModel(
                [("player_id", ASCENDING), ("skill_id", ASCENDING)],
                unique=True,
            ),
        ]


def insert_data():
    brazil = Country(name="Brazil").create()
    argentina = Country(name="Argentina").create()
    france = Country(name="France").create()

    catching = Skill(name="Catching")
    running = Skill(name="Running")
    kicking = Skill(name="Kicking")

    pele = Player(name="Pelé", country_id=brazil.id).create()
    PlayerSkill(player_id=pele.id, skill_id=catching.id, rating=49).create()
    PlayerSkill(player_id=pele.id, skill_id=kicking.id, rating=49).create()

    maradona = Player(name="Diego Maradona", country_id=argentina.id).create()
    PlayerSkill(player_id=maradona.id, skill_id=catching.id, rating=48).create()
    PlayerSkill(player_id=maradona.id, skill_id=kicking.id, rating=49).create()

    zidane = Player(name="Zinedine Zidane", country_id=france.id).create()
    PlayerSkill(player_id=zidane.id, skill_id=running.id, rating=42).create()
    PlayerSkill(player_id=zidane.id, skill_id=kicking.id, rating=42).create()


def read_data():
    pele = Player.get({Player.name: "Pelé"})
    skills = list(PlayerSkill.find({"player_id": pele.id}))

    print(pele)
    print(skills)


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


def main():
    configuration()
    insert_data()
    read_data()


if __name__ == "__main__":
    main()

After running the read_data function, the console should print:

Player(id=ObjectId('id'), name='Pelé', country_id=ObjectId('id'), skills=[EmbeddedSkill(skill_id=ObjectId('id'), rating=49), EmbeddedSkill(skill_id=ObjectId('id'), rating=49)], _id=ObjectId('id'))

Run Full Code

Run the code and check the MongoDB document viewer to see the impact.

import os
from typing import Optional

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


class Country(Document):
    name: str


class Skill(Document):
    name: str


class EmbeddedSkill(BaseModel):
    skill_id: ODMObjectId
    rating: int = Field(default=0, ge=1, le=100)


class Player(Document):
    name: str
    country_id: ODMObjectId
    skills: list[EmbeddedSkill] = []

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

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


def insert_data():
    brazil = Country(name="Brazil").create()
    argentina = Country(name="Argentina").create()
    france = Country(name="France").create()

    catching = Skill(name="Catching")
    running = Skill(name="Running")
    kicking = Skill(name="Kicking")

    Player(
        name="Pelé",
        country_id=brazil.id,
        skills=[
            EmbeddedSkill(skill_id=catching.id, rating=49),
            EmbeddedSkill(skill_id=kicking.id, rating=49),
        ],
    ).create()
    Player(
        name="Diego Maradona",
        country_id=argentina.id,
        skills=[
            EmbeddedSkill(skill_id=catching.id, rating=48),
            EmbeddedSkill(skill_id=kicking.id, rating=49),
        ],
    ).create()
    Player(
        name="Zinedine Zidane",
        country_id=france.id,
        skills=[
            EmbeddedSkill(skill_id=running.id, rating=42),
            EmbeddedSkill(skill_id=kicking.id, rating=42),
        ],
    ).create()


def read_data():
    pele = Player.get({Player.name: "Pelé"})
    print(pele)


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


def main():
    configuration()
    insert_data()
    read_data()


if __name__ == "__main__":
    main()