Pydantic Examples¶

"""
This example demonstrates pydantic serialisation
"""

from tortoise import Tortoise, fields, run_async
from tortoise.contrib.pydantic import pydantic_model_creator, pydantic_queryset_creator
from tortoise.models import Model


class Tournament(Model):
    id = fields.IntField(primary_key=True)
    name = fields.TextField()
    created_at = fields.DatetimeField(auto_now_add=True)

    events: fields.ReverseRelation["Event"]

    class Meta:
        ordering = ["name"]


class Event(Model):
    id = fields.IntField(primary_key=True)
    name = fields.TextField()
    created_at = fields.DatetimeField(auto_now_add=True)
    tournament: fields.ForeignKeyNullableRelation[Tournament] = fields.ForeignKeyField(
        "models.Tournament", related_name="events", null=True
    )
    participants: fields.ManyToManyRelation["Team"] = fields.ManyToManyField(
        "models.Team", related_name="events", through="event_team"
    )
    address: fields.OneToOneNullableRelation["Address"]

    class Meta:
        ordering = ["name"]


class Address(Model):
    city = fields.CharField(max_length=64)
    street = fields.CharField(max_length=128)
    created_at = fields.DatetimeField(auto_now_add=True)

    event: fields.OneToOneRelation[Event] = fields.OneToOneField(
        "models.Event", on_delete=fields.OnDelete.CASCADE, related_name="address", primary_key=True
    )

    class Meta:
        ordering = ["city"]


class Team(Model):
    id = fields.IntField(primary_key=True)
    name = fields.TextField()
    created_at = fields.DatetimeField(auto_now_add=True)

    events: fields.ManyToManyRelation[Event]

    class Meta:
        ordering = ["name"]


async def run():
    await Tortoise.init(db_url="sqlite://:memory:", modules={"models": ["__main__"]})
    await Tortoise.generate_schemas()
    Event_Pydantic = pydantic_model_creator(Event)
    Event_Pydantic_List = pydantic_queryset_creator(Event)
    Tournament_Pydantic = pydantic_model_creator(Tournament)
    Team_Pydantic = pydantic_model_creator(Team)

    # print(Event_Pydantic_List.schema_json(indent=4))
    # print(Event_Pydantic.schema_json(indent=4))
    # print(Tournament_Pydantic.schema_json(indent=4))
    # print(Team_Pydantic.schema_json(indent=4))

    tournament = await Tournament.create(name="New Tournament")
    tournament2 = await Tournament.create(name="Old Tournament")
    await Event.create(name="Empty")
    event = await Event.create(name="Test", tournament=tournament)
    event2 = await Event.create(name="TestLast", tournament=tournament)
    event3 = await Event.create(name="Test2", tournament=tournament2)
    await Address.create(city="Santa Monica", street="Ocean", event=event)
    await Address.create(city="Somewhere Else", street="Lane", event=event2)
    team1 = await Team.create(name="Onesies")
    team2 = await Team.create(name="T-Shirts")
    team3 = await Team.create(name="Alternates")
    await event.participants.add(team1, team2, team3)
    await event2.participants.add(team1, team2)
    await event3.participants.add(team1, team3)

    p = await Event_Pydantic.from_tortoise_orm(await Event.get(name="Test"))
    print("One Event:", p.model_dump_json(indent=4))

    p = await Tournament_Pydantic.from_tortoise_orm(await Tournament.get(name="New Tournament"))
    print("One Tournament:", p.model_dump_json(indent=4))

    p = await Team_Pydantic.from_tortoise_orm(await Team.get(name="Onesies"))
    print("One Team:", p.model_dump_json(indent=4))

    pl = await Event_Pydantic_List.from_queryset(Event.filter(address__event_id__isnull=True))
    print("All Events without addresses:", pl.model_dump_json(indent=4))


if __name__ == "__main__":
    run_async(run())

"""
This example demonstrates pydantic serialisation, and how to use early partial init.
"""

from tortoise import Tortoise, fields
from tortoise.contrib.pydantic import pydantic_model_creator
from tortoise.models import Model


class Tournament(Model):
    id = fields.IntField(primary_key=True)
    name = fields.TextField()
    created_at = fields.DatetimeField(auto_now_add=True)

    events: fields.ReverseRelation["Event"]

    class Meta:
        ordering = ["name"]


class Event(Model):
    id = fields.IntField(primary_key=True)
    name = fields.TextField()
    created_at = fields.DatetimeField(auto_now_add=True)
    tournament: fields.ForeignKeyNullableRelation[Tournament] = fields.ForeignKeyField(
        "models.Tournament", related_name="events", null=True
    )

    class Meta:
        ordering = ["name"]


Event_TooEarly = pydantic_model_creator(Event)
print("Relations are missing if models not initialized:")
print(Event_TooEarly.schema_json(indent=4))


Tortoise.init_models(["__main__"], "models")

Event_Pydantic = pydantic_model_creator(Event)
print("\nRelations are now present:")
print(Event_Pydantic.schema_json(indent=4))

# Now we can use the pydantic model early if needed

"""
This example demonstrates pydantic serialisation of a recursively cycled model.
"""

from tortoise import Tortoise, fields, run_async
from tortoise.contrib.pydantic import pydantic_model_creator
from tortoise.exceptions import NoValuesFetched
from tortoise.models import Model


class Employee(Model):
    name = fields.CharField(max_length=50)

    manager: fields.ForeignKeyNullableRelation["Employee"] = fields.ForeignKeyField(
        "models.Employee", related_name="team_members", null=True
    )
    team_members: fields.ReverseRelation["Employee"]

    talks_to: fields.ManyToManyRelation["Employee"] = fields.ManyToManyField(
        "models.Employee", related_name="gets_talked_to"
    )
    gets_talked_to: fields.ManyToManyRelation["Employee"]

    def name_length(self) -> int:
        # Computes length of name
        # Note that this function needs to be annotated with a return type so that pydantic
        # can generate a valid schema
        return len(self.name)

    def team_size(self) -> int:
        """
        Computes team size.

        Note that this function needs to be annotated with a return type so that pydantic can
         generate a valid schema.

        Note that the pydantic serializer can't call async methods, but the tortoise helpers
         pre-fetch relational data, so that it is available before serialization. So we don't
         need to await the relation. We do however have to protect against the case where no
         prefetching was done, hence catching and handling the
         ``tortoise.exceptions.NoValuesFetched`` exception.
        """
        try:
            return len(self.team_members)
        except NoValuesFetched:
            return -1

    def not_annotated(self):
        # Never called due to no annotation!
        raise NotImplementedError("Not Done")

    class PydanticMeta:
        computed = ["name_length", "team_size", "not_annotated"]
        exclude = ["manager", "gets_talked_to"]
        allow_cycles = True
        max_recursion = 4


async def run():
    await Tortoise.init(db_url="sqlite://:memory:", modules={"models": ["__main__"]})
    await Tortoise.generate_schemas()

    Employee_Pydantic = pydantic_model_creator(Employee)
    # print(Employee_Pydantic.schema_json(indent=4))

    root = await Employee.create(name="Root")
    loose = await Employee.create(name="Loose")
    _1 = await Employee.create(name="1. First H1", manager=root)
    _2 = await Employee.create(name="2. Second H1", manager=root)
    _1_1 = await Employee.create(name="1.1. First H2", manager=_1)
    _1_1_1 = await Employee.create(name="1.1.1. First H3", manager=_1_1)
    _2_1 = await Employee.create(name="2.1. Second H2", manager=_2)
    _2_2 = await Employee.create(name="2.2. Third H2", manager=_2)

    await _1.talks_to.add(_2, _1_1_1, loose)
    await _2_1.gets_talked_to.add(_2_2, _1_1, loose)

    p = await Employee_Pydantic.from_tortoise_orm(await Employee.get(name="Root"))
    print(p.model_dump_json(indent=4))


if __name__ == "__main__":
    run_async(run())

"""
Pydantic tutorial 1

Here we introduce:
* Creating a Pydantic model from a Tortoise model
* Docstrings & doc-comments are used
* Evaluating the generated schema
* Simple serialisation with both .model_dump() and .model_dump_json()
"""

from tortoise import Tortoise, fields, run_async
from tortoise.contrib.pydantic import pydantic_model_creator
from tortoise.models import Model


class Tournament(Model):
    """
    This references a Tournament
    """

    id = fields.IntField(primary_key=True)
    name = fields.CharField(max_length=100)
    #: The date-time the Tournament record was created at
    created_at = fields.DatetimeField(auto_now_add=True)


Tournament_Pydantic = pydantic_model_creator(Tournament)
# Print JSON-schema
print(Tournament_Pydantic.schema_json(indent=4))


async def run():
    await Tortoise.init(db_url="sqlite://:memory:", modules={"models": ["__main__"]})
    await Tortoise.generate_schemas()

    # Create object
    tournament = await Tournament.create(name="New Tournament")
    # Serialise it
    tourpy = await Tournament_Pydantic.from_tortoise_orm(tournament)

    # As Python dict with Python objects (e.g. datetime)
    print(tourpy.model_dump())
    # As serialised JSON (e.g. datetime is ISO8601 string representation)
    print(tourpy.model_dump_json(indent=4))


if __name__ == "__main__":
    run_async(run())

"""
Pydantic tutorial 2

Here we introduce:
* Creating a list-model to serialise a queryset
* Default sorting is honoured
"""

from tortoise import Tortoise, fields, run_async
from tortoise.contrib.pydantic import pydantic_queryset_creator
from tortoise.models import Model


class Tournament(Model):
    """
    This references a Tournament
    """

    id = fields.IntField(primary_key=True)
    name = fields.CharField(max_length=100)
    #: The date-time the Tournament record was created at
    created_at = fields.DatetimeField(auto_now_add=True)

    class Meta:
        # Define the default ordering
        #  the pydantic serialiser will use this to order the results
        ordering = ["name"]


# Create a list of models for population from a queryset.
Tournament_Pydantic_List = pydantic_queryset_creator(Tournament)
# Print JSON-schema
print(Tournament_Pydantic_List.schema_json(indent=4))


async def run():
    await Tortoise.init(db_url="sqlite://:memory:", modules={"models": ["__main__"]})
    await Tortoise.generate_schemas()

    # Create objects
    await Tournament.create(name="New Tournament")
    await Tournament.create(name="Another")
    await Tournament.create(name="Last Tournament")

    # Serialise it
    tourpy = await Tournament_Pydantic_List.from_queryset(Tournament.all())

    # As Python dict with Python objects (e.g. datetime)
    # Note that the root element is 'root' that contains the root element.
    print(tourpy.model_dump())
    # As serialised JSON (e.g. datetime is ISO8601 string representation)
    print(tourpy.model_dump_json(indent=4))


if __name__ == "__main__":
    run_async(run())

"""
Pydantic tutorial 3

Here we introduce:
* Relationships
* Early model init
"""

from tortoise import Tortoise, fields, run_async
from tortoise.contrib.pydantic import pydantic_model_creator
from tortoise.models import Model


class Tournament(Model):
    """
    This references a Tournament
    """

    id = fields.IntField(primary_key=True)
    name = fields.CharField(max_length=100)
    #: The date-time the Tournament record was created at
    created_at = fields.DatetimeField(auto_now_add=True)


class Event(Model):
    """
    This references an Event in a Tournament
    """

    id = fields.IntField(primary_key=True)
    name = fields.CharField(max_length=100)
    created_at = fields.DatetimeField(auto_now_add=True)

    tournament: fields.ForeignKeyRelation[Tournament] = fields.ForeignKeyField(
        "models.Tournament", related_name="events", description="The Tournament this happens in"
    )


# Early model, does not include relations
Tournament_Pydantic_Early = pydantic_model_creator(Tournament)
# Print JSON-schema
print(Tournament_Pydantic_Early.schema_json(indent=4))


# Initialise model structure early. This does not init any database structures
Tortoise.init_models(["__main__"], "models")


# We now have a complete model
Tournament_Pydantic = pydantic_model_creator(Tournament)
# Print JSON-schema
print(Tournament_Pydantic.schema_json(indent=4))

# Note how both schema's don't follow relations back.
Event_Pydantic = pydantic_model_creator(Event)
# Print JSON-schema
print(Event_Pydantic.schema_json(indent=4))


async def run():
    await Tortoise.init(db_url="sqlite://:memory:", modules={"models": ["__main__"]})
    await Tortoise.generate_schemas()

    # Create objects
    tournament = await Tournament.create(name="New Tournament")
    event = await Event.create(name="The Event", tournament=tournament)

    # Serialise Tournament
    tourpy = await Tournament_Pydantic.from_tortoise_orm(tournament)

    # As serialised JSON
    print(tourpy.model_dump_json(indent=4))

    # Serialise Event
    eventpy = await Event_Pydantic.from_tortoise_orm(event)

    # As serialised JSON
    print(eventpy.model_dump_json(indent=4))


if __name__ == "__main__":
    run_async(run())

"""
Pydantic tutorial 4

Here we introduce:
* Configuring model creator via PydanticMeta class.
* Using callable functions to annotate extra data.
"""

from tortoise import Tortoise, fields, run_async
from tortoise.contrib.pydantic import pydantic_model_creator
from tortoise.exceptions import NoValuesFetched
from tortoise.models import Model


class Tournament(Model):
    """
    This references a Tournament
    """

    id = fields.IntField(primary_key=True)
    name = fields.CharField(max_length=100)
    created_at = fields.DatetimeField(auto_now_add=True)

    # It is useful to define the reverse relations manually so that type checking
    #  and auto completion work
    events: fields.ReverseRelation["Event"]

    def name_length(self) -> int:
        """
        Computes length of name
        """
        # Note that this function needs to be annotated with a return type so that pydantic
        #  can generate a valid schema
        return len(self.name)

    def events_num(self) -> int:
        """
        Computes team size.
        """
        # Note that this function needs to be annotated with a return type so that pydantic
        #  can generate a valid schema.

        # Note that the pydantic serializer can't call async methods, but the tortoise helpers
        #  pre-fetch relational data, so that it is available before serialization. So we don't
        #  need to await the relation. We do however have to protect against the case where no
        #  prefetching was done, hence catching and handling the
        #  ``tortoise.exceptions.NoValuesFetched`` exception
        try:
            return len(self.events)
        except NoValuesFetched:
            return -1

    class PydanticMeta:
        # Let's exclude the created timestamp
        exclude = ("created_at",)
        # Let's include two callables as computed columns
        computed = ("name_length", "events_num")


class Event(Model):
    """
    This references an Event in a Tournament
    """

    id = fields.IntField(primary_key=True)
    name = fields.CharField(max_length=100)
    created_at = fields.DatetimeField(auto_now_add=True)

    tournament: fields.ForeignKeyRelation[Tournament] = fields.ForeignKeyField(
        "models.Tournament", related_name="events", description="The Tournament this happens in"
    )

    class Meta:
        ordering = ["name"]

    class PydanticMeta:
        exclude = ("created_at",)


# Initialise model structure early. This does not init any database structures
Tortoise.init_models(["__main__"], "models")
Tournament_Pydantic = pydantic_model_creator(Tournament)


# Print JSON-schema
print(Tournament_Pydantic.schema_json(indent=4))


async def run():
    await Tortoise.init(db_url="sqlite://:memory:", modules={"models": ["__main__"]})
    await Tortoise.generate_schemas()

    # Create objects
    tournament = await Tournament.create(name="New Tournament")
    await Event.create(name="Event 1", tournament=tournament)
    await Event.create(name="Event 2", tournament=tournament)

    # Serialise Tournament
    tourpy = await Tournament_Pydantic.from_tortoise_orm(tournament)

    # As serialised JSON
    print(tourpy.model_dump_json(indent=4))


if __name__ == "__main__":
    run_async(run())