Simple Examples¶

"""
This example demonstrates most basic operations with single model
"""

from tortoise import fields, run_async
from tortoise.contrib.test import init_memory_sqlite
from tortoise.models import Model


class Event(Model):
    id = fields.IntField(primary_key=True)
    name = fields.TextField()
    datetime = fields.DatetimeField(null=True)

    class Meta:
        table = "event"

    def __str__(self):
        return self.name


@init_memory_sqlite
async def run() -> None:
    event = await Event.create(name="Test")
    await Event.filter(id=event.id).update(name="Updated name")

    print(await Event.filter(name="Updated name").first())
    # >>> Updated name

    await Event(name="Test 2").save()
    print(await Event.all().values_list("id", flat=True))
    # >>> [1, 2]
    print(await Event.all().values("id", "name"))
    # >>> [{'id': 1, 'name': 'Updated name'}, {'id': 2, 'name': 'Test 2'}]


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

"""
This example demonstrates most basic operations with single model
and a Table definition generation with comment support
"""

from tortoise import Tortoise, fields, run_async
from tortoise.models import Model


class Event(Model):
    id = fields.IntField(primary_key=True)
    name = fields.TextField(description="Name of the event that corresponds to an action")
    datetime = fields.DatetimeField(
        null=True, description="Datetime of when the event was generated"
    )

    class Meta:
        table = "event"
        table_description = "This table contains a list of all the example events"

    def __str__(self):
        return self.name


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

    event = await Event.create(name="Test")
    await Event.filter(id=event.id).update(name="Updated name")

    print(await Event.filter(name="Updated name").first())

    await Event(name="Test 2").save()
    print(await Event.all().values_list("id", flat=True))
    print(await Event.all().values("id", "name"))


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

from tortoise import Tortoise, fields, run_async
from tortoise.models import Model
from tortoise.query_utils import Prefetch


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

    events: fields.ReverseRelation["Event"]

    def __str__(self):
        return self.name


class Event(Model):
    id = fields.IntField(primary_key=True)
    name = fields.TextField()
    tournament: fields.ForeignKeyRelation[Tournament] = fields.ForeignKeyField(
        "models.Tournament", related_name="events"
    )
    participants: fields.ManyToManyRelation["Team"] = fields.ManyToManyField(
        "models.Team", related_name="events", through="event_team"
    )

    def __str__(self):
        return self.name


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

    events: fields.ManyToManyRelation[Event]

    def __str__(self):
        return self.name


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

    tournament = await Tournament.create(name="tournament")
    await Event.create(name="First", tournament=tournament)
    await Event.create(name="Second", tournament=tournament)
    tournament_with_filtered = (
        await Tournament.all()
        .prefetch_related(Prefetch("events", queryset=Event.filter(name="First")))
        .first()
    )
    print(tournament_with_filtered)
    print(await Tournament.first().prefetch_related("events"))

    tournament_with_filtered_to_attr = (
        await Tournament.all()
        .prefetch_related(
            Prefetch("events", queryset=Event.filter(name="First"), to_attr="to_attr_events_first"),
            Prefetch(
                "events", queryset=Event.filter(name="Second"), to_attr="to_attr_events_second"
            ),
        )
        .first()
    )
    print(tournament_with_filtered_to_attr.to_attr_events_first)
    print(tournament_with_filtered_to_attr.to_attr_events_second)


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

"""
This example demonstrates how you can use transactions with tortoise
"""

from tortoise import Tortoise, fields, run_async
from tortoise.exceptions import OperationalError
from tortoise.models import Model
from tortoise.transactions import atomic, in_transaction


class Event(Model):
    id = fields.IntField(primary_key=True)
    name = fields.TextField()

    class Meta:
        table = "event"

    def __str__(self):
        return self.name


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

    try:
        async with in_transaction() as connection:
            event = Event(name="Test")
            await event.save(using_db=connection)
            await Event.filter(id=event.id).using_db(connection).update(name="Updated name")
            saved_event = await Event.filter(name="Updated name").using_db(connection).first()
            await connection.execute_query("SELECT * FROM non_existent_table")
    except OperationalError:
        pass
    saved_event = await Event.filter(name="Updated name").first()
    print(saved_event)

    @atomic()
    async def bound_to_fall():
        event = await Event.create(name="Test")
        await Event.filter(id=event.id).update(name="Updated name")
        saved_event = await Event.filter(name="Updated name").first()
        print(saved_event.name)
        raise OperationalError()

    try:
        await bound_to_fall()
    except OperationalError:
        pass
    saved_event = await Event.filter(name="Updated name").first()
    print(saved_event)


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

from tortoise import Tortoise, fields, run_async
from tortoise.expressions import Q
from tortoise.functions import Coalesce, Count, Length, Lower, Min, Sum, Trim, Upper
from tortoise.models import Model


class Tournament(Model):
    id = fields.IntField(primary_key=True)
    name = fields.TextField()
    desc = fields.TextField(null=True)

    events: fields.ReverseRelation["Event"]

    def __str__(self):
        return self.name


class Event(Model):
    id = fields.IntField(primary_key=True)
    name = fields.TextField()
    tournament: fields.ForeignKeyRelation[Tournament] = fields.ForeignKeyField(
        "models.Tournament", related_name="events"
    )
    participants: fields.ManyToManyRelation["Team"] = fields.ManyToManyField(
        "models.Team", related_name="events", through="event_team"
    )

    def __str__(self):
        return self.name


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

    events: fields.ManyToManyRelation[Event]

    def __str__(self):
        return self.name


async def run():
    await Tortoise.init(db_url="sqlite://:memory:", modules={"models": ["__main__"]})
    await Tortoise.generate_schemas()
    tournament = await Tournament.create(name="New Tournament", desc="great")
    await tournament.save()
    await Tournament.create(name="Second tournament")
    await Tournament.create(name=" final tournament ")
    await Event(name="Without participants", tournament_id=tournament.id).save()
    event = Event(name="Test", tournament_id=tournament.id)
    await event.save()
    participants = []
    for i in range(2):
        team = Team(name=f"Team {(i + 1)}")
        await team.save()
        participants.append(team)
    await event.participants.add(participants[0], participants[1])
    await event.participants.add(participants[0], participants[1])

    print(await Tournament.all().annotate(events_count=Count("events")).filter(events_count__gte=1))
    print(
        await Tournament.all()
        .annotate(events_count_with_filter=Count("events", _filter=Q(name="New Tournament")))
        .filter(events_count_with_filter__gte=1)
    )

    print(await Event.filter(id=event.id).first().annotate(lowest_team_id=Min("participants__id")))

    print(await Tournament.all().annotate(events_count=Count("events")).order_by("events_count"))

    print(await Event.all().annotate(tournament_test_id=Sum("tournament__id")).first())

    print(
        await Tournament.annotate(clean_description=Coalesce("desc", "")).filter(
            clean_description=""
        )
    )

    print(
        await Tournament.annotate(trimmed_name=Trim("name")).filter(trimmed_name="final tournament")
    )

    print(
        await Tournament.annotate(name_len=Length("name")).filter(
            name_len__gt=len("New Tournament")
        )
    )

    print(await Tournament.annotate(name_lo=Lower("name")).filter(name_lo="new tournament"))
    print(await Tournament.annotate(name_lo=Upper("name")).filter(name_lo="NEW TOURNAMENT"))


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

from tortoise import Model, Tortoise, fields, run_async
from tortoise.functions import Avg, Count, Sum


class Author(Model):
    name = fields.CharField(max_length=255)


class Book(Model):
    name = fields.CharField(max_length=255)
    author: fields.ForeignKeyRelation[Author] = fields.ForeignKeyField(
        "models.Author", related_name="books"
    )
    rating = fields.FloatField()


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

    a1 = await Author.create(name="author1")
    a2 = await Author.create(name="author2")
    for i in range(10):
        await Book.create(name=f"book{i}", author=a1, rating=i)
    for i in range(5):
        await Book.create(name=f"book{i}", author=a2, rating=i)

    ret = await Book.annotate(count=Count("id")).group_by("author_id").values("author_id", "count")
    print(ret)
    # >>> [{'author_id': 1, 'count': 10}, {'author_id': 2, 'count': 5}]

    ret = (
        await Book.annotate(count=Count("id"))
        .filter(count__gt=6)
        .group_by("author_id")
        .values("author_id", "count")
    )
    print(ret)
    # >>> [{'author_id': 1, 'count': 10}]

    ret = await Book.annotate(sum=Sum("rating")).group_by("author_id").values("author_id", "sum")
    print(ret)
    # >>> [{'author_id': 1, 'sum': 45.0}, {'author_id': 2, 'sum': 10.0}]

    ret = (
        await Book.annotate(sum=Sum("rating"))
        .filter(sum__gt=11)
        .group_by("author_id")
        .values("author_id", "sum")
    )
    print(ret)
    # >>> [{'author_id': 1, 'sum': 45.0}]

    ret = await Book.annotate(avg=Avg("rating")).group_by("author_id").values("author_id", "avg")
    print(ret)
    # >>> [{'author_id': 1, 'avg': 4.5}, {'author_id': 2, 'avg': 2.0}]

    ret = (
        await Book.annotate(avg=Avg("rating"))
        .filter(avg__gt=3)
        .group_by("author_id")
        .values("author_id", "avg")
    )
    print(ret)
    # >>> [{'author_id': 1, 'avg': 4.5}]

    # and use .values_list()
    ret = (
        await Book.annotate(count=Count("id"))
        .group_by("author_id")
        .values_list("author_id", "count")
    )
    print(ret)
    # >>> [(1, 10), (2, 5)]

    # group by with join
    ret = (
        await Book.annotate(count=Count("id"))
        .group_by("author__name")
        .values("author__name", "count")
    )
    print(ret)
    # >>> [{"author__name": "author1", "count": 10}, {"author__name": "author2", "count": 5}]


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

"""
This example demonstrates SQL Schema generation for each DB type supported.
"""

from tortoise import Tortoise, connections, fields, run_async
from tortoise.models import Model
from tortoise.utils import get_schema_sql


class Tournament(Model):
    id = fields.IntField(primary_key=True)
    name = fields.CharField(max_length=255, description="Tournament name", db_index=True)
    created = fields.DatetimeField(auto_now_add=True, description="Created datetime")

    events: fields.ReverseRelation["Event"]

    class Meta:
        table_description = "What Tournaments we have"


class Event(Model):
    id = fields.IntField(primary_key=True, description="Event ID")
    name = fields.CharField(max_length=255, unique=True)
    tournament: fields.ForeignKeyRelation[Tournament] = fields.ForeignKeyField(
        "models.Tournament", related_name="events", description="FK to tournament"
    )
    participants: fields.ManyToManyRelation["Team"] = fields.ManyToManyField(
        "models.Team",
        related_name="events",
        through="event_team",
        description="How participants relate",
    )
    modified = fields.DatetimeField(auto_now=True)
    prize = fields.DecimalField(max_digits=10, decimal_places=2, null=True)
    token = fields.CharField(max_length=100, description="Unique token", unique=True)

    class Meta:
        table_description = "This table contains a list of all the events"


class Team(Model):
    name = fields.CharField(max_length=50, primary_key=True, description="The TEAM name (and PK)")

    events: fields.ManyToManyRelation[Event]

    class Meta:
        table_description = "The TEAMS!"


async def run():
    print("SQLite:\n")
    await Tortoise.init(db_url="sqlite://:memory:", modules={"models": ["__main__"]})
    sql = get_schema_sql(connections.get("default"), safe=False)
    print(sql)

    print("\n\nMySQL:\n")
    await Tortoise.init(db_url="mysql://root:@127.0.0.1:3306/", modules={"models": ["__main__"]})
    sql = get_schema_sql(connections.get("default"), safe=False)
    print(sql)

    print("\n\nPostgreSQL:\n")
    await Tortoise.init(
        db_url="postgres://postgres:@127.0.0.1:5432/", modules={"models": ["__main__"]}
    )
    sql = get_schema_sql(connections.get("default"), safe=False)
    print(sql)


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

"""
This example demonstrates how you can use Tortoise if you have to
separate databases

Disclaimer: Although it allows to use two databases, you can't
use relations between two databases

Key notes of this example is using db_route for Tortoise init
and explicitly declaring model apps in class Meta
"""

from tortoise import Tortoise, connections, fields, run_async
from tortoise.exceptions import OperationalError
from tortoise.models import Model


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

    def __str__(self):
        return self.name

    class Meta:
        app = "tournaments"


class Event(Model):
    id = fields.IntField(primary_key=True)
    name = fields.TextField()
    tournament_id = fields.IntField()
    # Here we make link to events.Team, not models.Team
    participants: fields.ManyToManyRelation["Team"] = fields.ManyToManyField(
        "events.Team", related_name="events", through="event_team"
    )

    def __str__(self):
        return self.name

    class Meta:
        app = "events"


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

    event_team: fields.ManyToManyRelation[Event]

    def __str__(self):
        return self.name

    class Meta:
        app = "events"


async def run():
    await Tortoise.init(
        {
            "connections": {
                "first": {
                    "engine": "tortoise.backends.sqlite",
                    "credentials": {"file_path": "example.sqlite3"},
                },
                "second": {
                    "engine": "tortoise.backends.sqlite",
                    "credentials": {"file_path": "example1.sqlite3"},
                },
            },
            "apps": {
                "tournaments": {"models": ["__main__"], "default_connection": "first"},
                "events": {"models": ["__main__"], "default_connection": "second"},
            },
        }
    )
    await Tortoise.generate_schemas()
    client = connections.get("first")
    second_client = connections.get("second")

    tournament = await Tournament.create(name="Tournament")
    await Event(name="Event", tournament_id=tournament.id).save()

    try:
        await client.execute_query('SELECT * FROM "event"')
    except OperationalError:
        print("Expected it to fail")
    results = await second_client.execute_query('SELECT * FROM "event"')
    print(results)


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

"""
This example shows some more complex querying

Key points are filtering by related names and using Q objects
"""

from tortoise import Tortoise, fields, run_async
from tortoise.expressions import Q
from tortoise.models import Model


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

    events: fields.ReverseRelation["Event"]

    def __str__(self):
        return self.name


class Event(Model):
    id = fields.IntField(primary_key=True)
    name = fields.TextField()
    tournament: fields.ForeignKeyRelation[Tournament] = fields.ForeignKeyField(
        "models.Tournament", related_name="events"
    )
    participants: fields.ManyToManyRelation["Team"] = fields.ManyToManyField(
        "models.Team", related_name="events", through="event_team"
    )

    def __str__(self):
        return self.name


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

    events: fields.ManyToManyRelation[Event]

    def __str__(self):
        return self.name


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

    tournament = Tournament(name="Tournament")
    await tournament.save()

    second_tournament = Tournament(name="Tournament 2")
    await second_tournament.save()

    event_first = Event(name="1", tournament=tournament)
    await event_first.save()
    event_second = await Event.create(name="2", tournament=second_tournament)
    await Event.create(name="3", tournament=tournament)
    await Event.create(name="4", tournament=second_tournament)

    await Event.filter(tournament=tournament)

    team_first = Team(name="First")
    await team_first.save()
    team_second = Team(name="Second")
    await team_second.save()

    await team_first.events.add(event_first)
    await event_second.participants.add(team_second)

    print(
        await Event.filter(Q(id__in=[event_first.id, event_second.id]) | Q(name="3"))
        .filter(participants__not=team_second.id)
        .order_by("tournament__id")
        .distinct()
    )

    print(await Team.filter(events__tournament_id=tournament.id).order_by("-events__name"))
    print(
        await Tournament.filter(events__name__in=["1", "3"])
        .order_by("-events__participants__name")
        .distinct()
    )

    print(await Team.filter(name__icontains="CON"))

    print(await Tournament.filter(events__participants__name__startswith="Fir"))
    print(await Tournament.filter(id__icontains=1).count())


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

"""
This example shows how relations between models work.

Key points in this example are use of ForeignKeyField and ManyToManyField
to declare relations and use of .prefetch_related() and .fetch_related()
to get this related objects
"""

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


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

    events: fields.ReverseRelation["Event"]

    def __str__(self):
        return self.name


class Event(Model):
    id = fields.IntField(primary_key=True)
    name = fields.TextField()
    tournament: fields.ForeignKeyRelation[Tournament] = fields.ForeignKeyField(
        "models.Tournament", related_name="events"
    )
    participants: fields.ManyToManyRelation["Team"] = fields.ManyToManyField(
        "models.Team", related_name="events", through="event_team"
    )

    def __str__(self):
        return self.name


class Address(Model):
    city = fields.CharField(max_length=64)
    street = fields.CharField(max_length=128)

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

    def __str__(self):
        return f"Address({self.city}, {self.street})"


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

    events: fields.ManyToManyRelation[Event]

    def __str__(self):
        return self.name


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

    tournament = Tournament(name="New Tournament")
    await tournament.save()
    await Event(name="Without participants", tournament_id=tournament.id).save()
    event = Event(name="Test", tournament_id=tournament.id)
    await event.save()

    await Address.create(city="Santa Monica", street="Ocean", event=event)

    participants = []
    for i in range(2):
        team = Team(name=f"Team {(i + 1)}")
        await team.save()
        participants.append(team)
    await event.participants.add(participants[0], participants[1])
    await event.participants.add(participants[0], participants[1])

    try:
        for team in event.participants:
            print(team.id)
    except NoValuesFetched:
        pass

    async for team in event.participants:
        print(team.id)

    for team in event.participants:
        print(team.id)

    print(
        await Event.filter(participants=participants[0].id).prefetch_related(
            "participants", "tournament"
        )
    )
    print(await participants[0].fetch_related("events"))

    print(await Team.fetch_for_list(participants, "events"))

    print(await Team.filter(events__tournament__id=tournament.id))

    print(await Event.filter(tournament=tournament))

    print(
        await Tournament.filter(events__name__in=["Test", "Prod"])
        .order_by("-events__participants__name")
        .distinct()
    )

    print(await Event.filter(id=event.id).values("id", "name", tournament="tournament__name"))

    print(await Event.filter(id=event.id).values_list("id", "participants__name"))

    print(await Address.filter(event=event).first())

    event_reload1 = await Event.filter(id=event.id).first()
    print(await event_reload1.address)

    event_reload2 = await Event.filter(id=event.id).prefetch_related("address").first()
    print(event_reload2.address)


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

"""
This example shows how relations between models especially unique field work.

Key points in this example are use of ForeignKeyField and OneToOneField has to_field.
For other basic parts, it is the same as relation example.
"""

from tortoise import Tortoise, fields, run_async
from tortoise.models import Model
from tortoise.query_utils import Prefetch


class School(Model):
    uuid = fields.UUIDField(primary_key=True)
    name = fields.TextField()
    id = fields.IntField(unique=True)

    students: fields.ReverseRelation["Student"]
    principal: fields.ReverseRelation["Principal"]


class Student(Model):
    id = fields.IntField(primary_key=True)
    name = fields.TextField()
    school: fields.ForeignKeyRelation[School] = fields.ForeignKeyField(
        "models.School", related_name="students", to_field="id"
    )


class Principal(Model):
    id = fields.IntField(primary_key=True)
    name = fields.TextField()
    school: fields.OneToOneRelation[School] = fields.OneToOneField(
        "models.School", on_delete=fields.OnDelete.CASCADE, related_name="principal", to_field="id"
    )


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

    school1 = await School.create(id=1024, name="School1")
    student1 = await Student.create(name="Sang-Heon Jeon1", school_id=school1.id)

    student_schools = await Student.filter(name="Sang-Heon Jeon1").values("name", "school__name")
    print(student_schools[0])

    await Student.create(name="Sang-Heon Jeon2", school=school1)
    school_with_filtered = (
        await School.all()
        .prefetch_related(Prefetch("students", queryset=Student.filter(name="Sang-Heon Jeon1")))
        .first()
    )
    school_without_filtered = await School.first().prefetch_related("students")
    print(len(school_with_filtered.students))
    print(len(school_without_filtered.students))

    school2 = await School.create(id=2048, name="School2")
    await Student.all().update(school=school2)
    student = await Student.first()
    print(student.school_id)

    await Student.filter(id=student1.id).update(school=school1)
    schools = await School.all().order_by("students__name")
    print([school.name for school in schools])

    fetched_principal = await Principal.create(name="Sang-Heon Jeon3", school=school1)
    print(fetched_principal.name)
    fetched_school = await School.filter(name="School1").prefetch_related("principal").first()
    print(fetched_school.name)


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

"""
This example shows how self-referential (recursive) relations work.

Key points in this example are:
* Use of ForeignKeyField that refers to self
* To pass in the (optional) parent node at creation
* To use async iterator to fetch children
* To use .fetch_related(…) to emulate sync behaviour
* That insert-order gets preserved for ForeignFields, but not ManyToManyFields
"""

from tortoise import Tortoise, fields, run_async
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 __str__(self):
        return self.name

    async def full_hierarchy__async_for(self, level=0):
        """
        Demonstrates ``async for` to fetch relations

        An async iterator will fetch the relationship on-demand.
        """
        text = [
            "{}{} (to: {}) (from: {})".format(
                level * "  ",
                self,
                ", ".join([str(val) async for val in self.talks_to]),
                ", ".join([str(val) async for val in self.gets_talked_to]),
            )
        ]
        async for member in self.team_members:
            text.append(await member.full_hierarchy__async_for(level + 1))
        return "\n".join(text)

    async def full_hierarchy__fetch_related(self, level=0):
        """
        Demonstrates ``await .fetch_related`` to fetch relations

        On prefetching the data, the relationship files will contain a regular list.

        This is how one would get relations working on sync serialization/templating frameworks.
        """
        await self.fetch_related("team_members", "talks_to", "gets_talked_to")
        text = [
            "{}{} (to: {}) (from: {})".format(
                level * "  ",
                self,
                ", ".join([str(val) for val in self.talks_to]),
                ", ".join([str(val) for val in self.gets_talked_to]),
            )
        ]
        for member in self.team_members:
            text.append(await member.full_hierarchy__fetch_related(level + 1))
        return "\n".join(text)


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

    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)

    # Evaluated off creation objects
    print(await loose.full_hierarchy__fetch_related())
    print(await root.full_hierarchy__async_for())
    print(await root.full_hierarchy__fetch_related())

    # Evaluated off new objects → Result is identical
    root2 = await Employee.get(name="Root")
    loose2 = await Employee.get(name="Loose")
    print(await loose2.full_hierarchy__fetch_related())
    print(await root2.full_hierarchy__async_for())
    print(await root2.full_hierarchy__fetch_related())


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

from enum import Enum, IntEnum

from tortoise import Tortoise, fields, run_async
from tortoise.models import Model


class Service(IntEnum):
    python_programming = 1
    database_design = 2
    system_administration = 3


class Currency(str, Enum):
    HUF = "HUF"
    EUR = "EUR"
    USD = "USD"


class EnumFields(Model):
    service: Service = fields.IntEnumField(Service)
    currency: Currency = fields.CharEnumField(Currency, default=Currency.HUF)


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

    obj0 = await EnumFields.create(service=Service.python_programming, currency=Currency.USD)
    # also you can use valid int and str value directly
    await EnumFields.create(service=1, currency="USD")

    try:
        # invalid enum value will raise ValueError
        await EnumFields.create(service=4, currency="XXX")
    except ValueError:
        print("Value is invalid")

    await EnumFields.filter(pk=obj0.pk).update(
        service=Service.database_design, currency=Currency.HUF
    )
    # also you can use valid int and str value directly
    await EnumFields.filter(pk=obj0.pk).update(service=2, currency="HUF")


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

"""
This example demonstrates model signals usage
"""

from typing import List, Optional, Type

from tortoise import BaseDBAsyncClient, Tortoise, fields, run_async
from tortoise.models import Model
from tortoise.signals import post_delete, post_save, pre_delete, pre_save


class Signal(Model):
    id = fields.IntField(primary_key=True)
    name = fields.TextField()

    class Meta:
        table = "signal"

    def __str__(self):
        return self.name


@pre_save(Signal)
async def signal_pre_save(
    sender: "Type[Signal]", instance: Signal, using_db, update_fields
) -> None:
    print(sender, instance, using_db, update_fields)


@post_save(Signal)
async def signal_post_save(
    sender: "Type[Signal]",
    instance: Signal,
    created: bool,
    using_db: "Optional[BaseDBAsyncClient]",
    update_fields: List[str],
) -> None:
    print(sender, instance, using_db, created, update_fields)


@pre_delete(Signal)
async def signal_pre_delete(
    sender: "Type[Signal]", instance: Signal, using_db: "Optional[BaseDBAsyncClient]"
) -> None:
    print(sender, instance, using_db)


@post_delete(Signal)
async def signal_post_delete(
    sender: "Type[Signal]", instance: Signal, using_db: "Optional[BaseDBAsyncClient]"
) -> None:
    print(sender, instance, using_db)


async def run():
    await Tortoise.init(db_url="sqlite://:memory:", modules={"models": ["__main__"]})
    await Tortoise.generate_schemas()
    # pre_save,post_save will be send
    signal = await Signal.create(name="Signal")
    signal.name = "Signal_Save"

    # pre_save,post_save will be send
    await signal.save(update_fields=["name"])

    # pre_delete,post_delete will be send
    await signal.delete()


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

"""
This example demonstrates executing manual SQL queries
"""

from tortoise import Tortoise, connections, fields, run_async
from tortoise.models import Model
from tortoise.transactions import in_transaction


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


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

    # Need to get a connection. Unless explicitly specified, the name should be 'default'
    conn = connections.get("default")

    # Now we can execute queries in the normal autocommit mode
    await conn.execute_query("INSERT INTO event (name) VALUES ('Foo')")

    # You can also you parameters, but you need to use the right param strings for each dialect
    await conn.execute_query("INSERT INTO event (name) VALUES (?)", ["Bar"])

    # To do a transaction you'd need to use the in_transaction context manager
    async with in_transaction("default") as tconn:
        await tconn.execute_query("INSERT INTO event (name) VALUES ('Moo')")
        # Unless an exception happens it should commit automatically

    # This transaction is rolled back
    async with in_transaction("default") as tconn:
        await tconn.execute_query("INSERT INTO event (name) VALUES ('Sheep')")
        # Rollback to fail transaction
        await tconn.rollback()

    # Consider using execute_query_dict to get return values as a dict
    val = await conn.execute_query_dict("SELECT * FROM event")
    print(val)
    # Note that the result doesn't contain the rolled-back "Sheep" entry.


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

"""
This example to use router to implement read/write separation
"""

from typing import Type

from tortoise import Tortoise, fields, run_async
from tortoise.models import Model


class Event(Model):
    id = fields.IntField(primary_key=True)
    name = fields.TextField()
    datetime = fields.DatetimeField(null=True)

    class Meta:
        table = "event"

    def __str__(self):
        return self.name


class Router:
    def db_for_read(self, model: Type[Model]):
        return "slave"

    def db_for_write(self, model: Type[Model]):
        return "master"


async def run():
    config = {
        "connections": {"master": "sqlite:///tmp/test.db", "slave": "sqlite:///tmp/test.db"},
        "apps": {
            "models": {
                "models": ["__main__"],
                "default_connection": "master",
            }
        },
        "routers": ["__main__.Router"],
        "use_tz": False,
        "timezone": "UTC",
    }
    await Tortoise.init(config=config)
    await Tortoise.generate_schemas()
    # this will use connection master
    event = await Event.create(name="Test")
    # this will use connection slave
    await Event.get(pk=event.pk)


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