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

"""
Pydantic computed fields example

Here we demonstrate:
* Nullable FK fields become Optional in the Pydantic schema (with ``"default": null``).
* Computed fields that access reverse relations work when the relation is included.
* Computed fields still work when the relation is excluded but manually prefetched.
* NoValuesFetched error propagation when a computed field accesses an unfetched relation.
* Graceful handling pattern using try/except inside the computed function.
"""

import json

from pydantic_core import PydanticSerializationError

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 Department(Model):
    id = fields.IntField(primary_key=True)
    name = fields.CharField(max_length=100)

    # Define reverse relation for type checking and auto completion
    employees: fields.ReverseRelation["Employee"]

    def employee_count(self) -> int:
        """
        Counts employees in the department.

        Uses try/except to gracefully handle the case where the relation has not been
        fetched, returning 0 instead of raising an error.
        """
        try:
            return len(self.employees)
        except (NoValuesFetched, AttributeError):
            return 0

    def employee_names(self) -> str:
        """
        Returns a comma-separated list of employee names.

        Does NOT handle NoValuesFetched -- demonstrates error propagation when the
        relation has not been fetched.
        """
        return ", ".join(e.name for e in self.employees)

    class PydanticMeta:
        computed = ("employee_count", "employee_names")


class Employee(Model):
    id = fields.IntField(primary_key=True)
    name = fields.CharField(max_length=100)

    department: fields.ForeignKeyNullableRelation[Department] = fields.ForeignKeyField(
        "models.Department", related_name="employees", null=True
    )


# Initialise model structure early so we can create pydantic models at module level
Tortoise.init_models(["__main__"], "models")


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

    # Create test data
    engineering = await Department.create(name="Engineering")
    await Employee.create(name="Alice", department=engineering)
    await Employee.create(name="Bob", department=engineering)
    await Employee.create(name="Charlie")  # no department (nullable FK)

    # ──────────────────────────────────────────────────────────────────────
    # Section 1: Nullable FK is Optional in the Pydantic schema
    #
    # A nullable ForeignKeyField generates a pydantic field with
    # "default": null that is NOT in "required".
    # ──────────────────────────────────────────────────────────────────────
    print("=" * 70)
    print("Section 1: Nullable FK is Optional")
    print("=" * 70)

    Employee_Pydantic = pydantic_model_creator(Employee)
    schema = Employee_Pydantic.model_json_schema()
    print(json.dumps(schema, indent=4))

    # The 'department' field has "default": null and is NOT in "required"
    required = schema.get("required", [])
    print(f"\nRequired fields: {required}")
    print("'department' in required:", "department" in required)

    dept_props = schema.get("properties", {}).get("department", {})
    has_default_null = dept_props.get("default") is None and "default" in dept_props
    print(f"'department' has default null: {has_default_null}")

    # ──────────────────────────────────────────────────────────────────────
    # Section 2: Computed field with included relation (auto-prefetched)
    #
    # When the 'employees' reverse relation is included in the pydantic
    # model, from_tortoise_orm() auto-prefetches it. Both computed fields
    # work because the relation data is available.
    # ──────────────────────────────────────────────────────────────────────
    print("\n" + "=" * 70)
    print("Section 2: Computed field with included relation")
    print("=" * 70)

    Department_Pydantic = pydantic_model_creator(Department)

    dept = await Department.get(name="Engineering")
    dept_pydantic = await Department_Pydantic.from_tortoise_orm(dept)
    print(dept_pydantic.model_dump_json(indent=4))

    # Both computed fields work because the relation was auto-prefetched
    print(f"\nemployee_count: {dept_pydantic.employee_count}")
    print(f"employee_names: {dept_pydantic.employee_names}")

    # ──────────────────────────────────────────────────────────────────────
    # Section 3: Computed field with excluded relation + manual prefetch
    #
    # The 'employees' relation is excluded from the pydantic model (so it
    # won't appear in the output), but we manually prefetch it before
    # serialization so the computed fields can still access the data.
    # ──────────────────────────────────────────────────────────────────────
    print("\n" + "=" * 70)
    print("Section 3: Excluded relation + manual prefetch")
    print("=" * 70)

    Department_Pydantic_NoEmployees = pydantic_model_creator(
        Department,
        name="Department_NoEmployees",
        exclude=("employees",),
    )

    dept = await Department.get(name="Engineering")
    # Manually prefetch the relation so computed fields can access it
    await dept.fetch_related("employees")
    dept_pydantic = await Department_Pydantic_NoEmployees.from_tortoise_orm(dept)
    print(dept_pydantic.model_dump_json(indent=4))

    print(f"\nemployee_count: {dept_pydantic.employee_count}")
    print(f"employee_names: {dept_pydantic.employee_names}")

    # ──────────────────────────────────────────────────────────────────────
    # Section 4: NoValuesFetched error propagation
    #
    # Same excluded model but WITHOUT manual prefetch. The employee_names()
    # computed field does not handle NoValuesFetched internally, so the
    # wrapper in creator.py re-raises with a descriptive message during
    # serialization.
    # ──────────────────────────────────────────────────────────────────────
    print("\n" + "=" * 70)
    print("Section 4: NoValuesFetched error propagation")
    print("=" * 70)

    dept = await Department.get(name="Engineering")
    dept_pydantic = await Department_Pydantic_NoEmployees.from_tortoise_orm(dept)
    try:
        # Serialization triggers the computed field, which raises NoValuesFetched
        dept_pydantic.model_dump_json(indent=4)
    except (PydanticSerializationError, NoValuesFetched) as e:
        print(f"Caught error during serialization: {e}")

    # ──────────────────────────────────────────────────────────────────────
    # Section 5: Graceful handling pattern
    #
    # A model with only the graceful computed field (employee_count) that
    # handles NoValuesFetched internally. Even without prefetching, it
    # returns 0 instead of crashing. Contrast with employee_names which
    # would fail in the same scenario.
    # ──────────────────────────────────────────────────────────────────────
    print("\n" + "=" * 70)
    print("Section 5: Graceful handling pattern")
    print("=" * 70)

    # Override PydanticMeta so only the graceful computed field is included
    class GracefulMeta:
        computed = ("employee_count",)

    Department_Pydantic_GracefulOnly = pydantic_model_creator(
        Department,
        name="Department_GracefulOnly",
        exclude=("employees",),
        meta_override=GracefulMeta,
    )

    dept = await Department.get(name="Engineering")
    # No manual prefetch -- employee_count() handles the exception internally
    dept_pydantic = await Department_Pydantic_GracefulOnly.from_tortoise_orm(dept)
    print(dept_pydantic.model_dump_json(indent=4))

    # employee_count returns 0 gracefully instead of crashing
    print(f"\nemployee_count (graceful, no prefetch): {dept_pydantic.employee_count}")
    print(
        "employee_names would raise NoValuesFetched in the same scenario, "
        "but employee_count handles it and returns 0."
    )


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

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