I have collected in one article everything you need to know about the tool and my experience working with it

How are things with FastAPI and SQLAlchemy?

SQLAlchemy (or Alchemy) is currently the de facto orm in the FastAPI stack. I'm sure there are many fans of this stack who enjoy working with a combination of these tools.

Relatively recently, FastAPI developer Tiangolo began actively promoting SQLModel in working with FastAPI, with which you can create both Pydantic classes and SQLAlchemy models at once. This reduces the amount of headaches, but many things in working with SQLAlchemy still remain inconvenient. I can’t say that I have much experience in Alchemy. Some of this may be incorrect, so please keep this in mind. Here are the negative points in working with this tool that I have noted for myself:

Forwarding db_session:

In order to work with SQLAlchemy, you have to pass db_session from the controller down through the abstraction layers, which is extremely inconvenient. For example, we have 4 abstraction layers, the bottom of which makes a request to the DB through Alchemy. At this point, you need to pass db_session from !!controller!! to this layer, which is extremely inconvenient. This is a constant argument to pass, which you need to remember.

Not very convenient query composition:

Personally, I don't like the style of writing queries in SQLAlchemy. You have to write a lot of unnecessary code to create a simple query. Yes, the level of control over them increases, but even the simplest ones can take up 3-4 lines. Let's say there are a lot of these queries in the service – then the code swells like yeast.

Not very convenient documentation:

Most likely, the problem is me, but I just get lost in the SQLAlchemy documentation. Skill issue, you say, and you might be right.

Building an additional layer for CRUD-like things:

In addition to business logic, you have to manually write functions/methods for working with models, for example: getting by pk, adding an object, deleting, etc.
Previously, the template from the FastAPI developers had CRUD, which could be hung on any model and basic operations would start working. However, even with this approach, you still have to write some custom selections in the CRUD of a specific model. More boilerplate code for the god of boilerplate code!

These points came directly from my experience using SQLAlchemy. All markers are different, so you can disagree with me and write about it in the comments. For me, these points were a hindrance when working with Alchemy. To get rid of this tool, I started looking for a replacement, which turned out to be TortoiseORM. We took it for a test run on the project, and eventually settled on it.

Description of Tortoise ORM

Tortoise ORM is an asynchronous library for working with a database. The first commit was uploaded to Github March 29, 2018. At the time of writing, Tortoise ORM has 4.3k stars on Github and is out 0.21.3 version, on the basis of which we will consider the library in the article. Under the hood, TortoiseORM uses pypika to create SQL Query.

Tortoise ORM will be very familiar, one might even say native to anyone who has worked with Django ORM before. The fact is that the developers created their ORM, inspired by Django ORM. However, Django ORM only comes bundled with Django and, although Django adds capabilities for asynchronous work, it still remains not the tool you choose when thinking about asynchronous.

The developers point out that Tortoise ORM is still a young library and changes may occur, breaking existing code when updating. It is worth keeping this in mind when using it in your projects.

Comparing Tortoise ORM Performance with Other ORM Libraries

The developers of Tortoise ORM have compared the performance of different ORM libraries for Python. As you can see from their benchmark, Tortoise ORM competes well with other ORM libraries, often trading places with Pony ORM. The developers are focused on performance, and this benchmark helped the creators find places that slow down the library.

Of course, questions arise regarding the fairness of such tests, since many variables must be taken into account for an objective comparison.

More detailed information about the benchmark can be found at Github

Basic Tortoise ORM Setup

First, you need to install the necessary packages:
pip install tortoise-orm

Additionally, you can install drivers for your database:

PostgreSQL:
pip install tortoise-orm[asyncpg]

MySQL:
pip install tortoise-orm[asyncmy]

Microsoft SQL Server / Oracle (not fully tested):
pip install tortoise-orm[asyncodbc]

Next, we set the ORM launch function, which we then call at the start of the application:

from tortoise import Tortoise

async def init():
    # Коннектимся к SQLite (подставляем свою бд)
    # Также обязательно указать модуль,
    # который содержит модели.
    await Tortoise.init(
        db_url="sqlite://db.sqlite3",
        modules={'models': ['app.models']}
    )
    # Генерируем схемы
    await Tortoise.generate_schemas()

After that we call init() at the start of our application – and we can use models in asynchronous functions.

It is important to remember that you need to close the connection to the database.
For this we use await Tortoise.close_connection()

In the future, we will use a similar code sketch to consider the capabilities of ORM:

from tortoise import Tortoise, fields, models, run_async


class Task(models.Model):
    id = fields.IntField(primary_key=True)
    name = fields.CharField(max_length=256)
    description = fields.CharField(max_length=500)
    date_created = fields.DatetimeField(auto_now_add=True)
    date_updated = fields.DatetimeField(auto_now=True)

    class Meta:
        table = "tasks"


async def main():
    await Tortoise.init(
        db_url="sqlite://db.sqlite3",
        # Модулем для моделей указываем __main__,
        # т.к. все модели для показа будем прописывать
        # именно тут
        modules={'models': ['__main__']},
    )
    await Tortoise.generate_schemas()

    task = await Task.create(
	    name="First task",
	    description="First task description"
	)
    print(task)
    # Output: <Task>
    print(task.name)
    # Output: First task

    task.name = "First task updated name"
    await task.save()
    print(task.name)
    # Output: First task updated name

    await Tortoise.close_connections()


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

Queryset Evaluation in Tortoise ORM

If you've worked with Django ORM, you know about Queryset Evaluation. This is the moment when your query is transformed into concrete objects. In Django, this is done implicitly, so it's important to know when your Queryset is transformed into objects.

In Tortoise ORM this is much easier, as async allowed us to make Queryset Evaluation explicit. The Queryset is built exactly until it is called await.

For example:

task_queryset = Task.filter(name="Task name")
print(task_queryset)
# Output: <Queryset object>

task = await task_queryset
print(task)
# Output: <Task>

Description of possible field types

Tortoise ORM supports the following field types (tortoise.fields):

BigIntField
BinaryField
BooleanField
CharEnumField
CharField
DateField
DatetimeField
TimeField
DecimalField
FloatField
IntEnumField
IntField
JSONField
SmallIntField
TextField
TimeDeltaField
UUIDField

There are also fields depending on the database used.
MySQL (tortoise.contrib.mysql.fields):

PostgreSQL (tortoise.contrib.postgres.fields):

You can also expand the fields by adding your own functionality. I will talk about this further in the article.

More about fields in the documentation: link

Description of simple operations

Tortoise ORM has a very simple API, which makes it a joy to work with. Again, for any developer who has worked with Django ORM, everything described below will be painfully familiar.

Let's briefly go over the API that Tortoise ORM provides.
More details in the documentation: link

As an example, we will use the following model:

class Task(models.Model):
    id = fields.IntField(primary_key=True)
    name = fields.CharField(max_length=256)
    description = fields.CharField(max_length=500)
    date_created = fields.DatetimeField(auto_now_add=True)
    date_updated = fields.DatetimeField(auto_now=True)

    class Meta:
        table = "tasks"

Creating an object:

task = Task(
    name="Task name",
    description="Task description",
)
await task.save()

However, there is no point in creating objects this way every time, because the developers made an alias for creation that does the same thing, but faster:

task = await Task.create(
	name="Task name",
	description="Task description"
)

Getting an object by field:

task = await Task.get(id=1)

If we call the code above, but there is no task with such id, we will get an exception, just like in Django.

In Django, such a case is often handled with two constructs:

try:
	task = Task.objects.get(id=1)
except Task.DoesNotExist:
	# код для обработки
	# Например:
	task = None

task = Task.objects.filter(id=1).first()

In Tortoise ORM, although it is possible, there is no need for such constructions, since there is a method get_or_none(**fields)

task = await Task.get_or_none(id=1)
if not task:
	# код для обработки

You can also get objects by multiple fields:

task = await Task.get_or_none(id=1, name="First task")

Through the method get_or_create(**fields) you can either get an object with the specified fields, or create a new one if no existing object meets the conditions:

# Возвращает tuple, в котором на первом индексе сам объект,
# а на втором булевый флаг, показывающий, был ли создан объект.
task, is_created = await Task.get_or_create(
	name="Task unique name",
	description="Task descrpition"
)
print(task)
# Output: <Task: 1>
print(is_created)
# Output: True

Getting a list of objects:
Just like in Django, you can get a list of objects via the method filter():

tasks = await Task.filter(name="Task name")

Tortoise ORM includes the following methods for selection:

filter(*args, **kwargs) — filter by specified field values
exclude(*args, **kwargs) — a filter that excludes objects that meet the selection conditions
all() – getting all objects of the model
first() – limits the selection to one object and returns it, instead of a list
annotate() — data annotation in the sample

As in Django, additional filters by fields can be specified in filter methods. For example, the following code will give us a list of objects whose names are included in the passed list:

task_names = ["Task 1", "Task 2", "Task 3"]
tasks = await Task.filter(name__in=task_names)

Tortoise ORM supports the following field filters:

not
in — checks whether the field value is included in the passed list
not_in
gte — greater than or equal to the passed value
gt — greater than the value passed
lte — less than or equal to the passed value
lt — less than the value passed
range — between the two values passed
isnull — is the field null
not_isnull — is the field null?
contains — the field contains a substring
icontains — contains case-insensitive
startswith — the field starts with a value
istartswith — startswith case-insensitive
endswith — the field ends with a value
iendswith — endswith case-insensitive
iexact — case insensitive comparison
search — full-text search

Examples:

task = await Task.create(
    name="First task",
    description="First task description",
)

for i in range(0, 5):
    await Task.create(name="s", description="s"+str(i))

t = await Task.filter(name__not="s")
print
# Output: [<Task: 1>]
    
task_descriptions = ['s0', 's1']
t2 = await Task.filter(description__in=task_descriptions)
print(t2)
# Output: [<Task: 2>, <Task: 3>]

t3 = await Task.filter(description__not_in=task_descriptions)
print(t3)
# Output: [<Task: 1>, <Task: 4>, <Task: 5>, <Task: 6>]

t4 = await Task.filter(id__gte=4)
print(t4)
# Output: [<Task: 4>, <Task: 5>, <Task: 6>]

t5 = await Task.filter(id__gt=4)
print(t5)
# Output: [<Task: 5>, <Task: 6>]

t6 = await Task.filter(id__lte=4)
print(t6)
# Output: [<Task: 1>, <Task: 2>, <Task: 3>, <Task: 4>]

t7 = await Task.filter(id__lt=4)
print(t7)
# Output: [<Task: 1>, <Task: 2>, <Task: 3>]

t8 = await Task.filter(id__range=[3, 5])
print(t8)
# Output: [<Task: 3>, <Task: 4>, <Task: 5>]

# gte, gt, lte, lt, range работают с Datetime полями
t4_dt = await Task.filter(
	date_created__gte=datetime.now() - timedelta(days=1)
)
print(t4_dt)
# Output: [<Task: 1>, <Task: 2>, <Task: 3>, <Task: 4>, <Task: 5>, <Task: 6>]

t6_dt = await Task.filter(
	date_created__lte=datetime.now() + timedelta(days=1)
)
print(t6_dt)
# Output: [<Task: 1>, <Task: 2>, <Task: 3>, <Task: 4>, <Task: 5>, <Task: 6>]

t8_dt = await Task.filter(date_created__range=['2024-06-08', '2024-06-12'])
print(t8_dt)
# Output: [<Task: 1>, <Task: 2>, <Task: 3>, <Task: 4>, <Task: 5>, <Task: 6>]
    
t9 = await Task.filter(name__contains="First")
print(t9)
# Output: [<Task: 1>]

t10 = await Task.filter(name__icontains="first")
print(t10)
# Output: [<Task: 1>]

t11 = await Task.filter(name__startswith="Fir")
print(t11)
# Output: [<Task: 1>]

t12 = await Task.filter(name__endswith="ask")
print(t12)
# Output: [<Task: 1>]

t13 = await Task.filter(name__iexact="first task")
print(t13)
# Output: [<Task: 1>]

There are also methods that transform a sample into a specific form:

count() — returns the number of objects, instead of the objects themselves.
distinct() — returns unique values from the sample. Only makes sense in combination with values() And values_list()because it filters incoming fields.
exists() – returns True/False for the presence of objects in the sample.
group_by() — returns a list of dictionaries or tuples with group by. Must be called before values() or values_list().
order_by() — sorts the selection by field/fields.
limit() — limits the sample to a given length.
offset() — offset for objects selected from the table.
values(*args, **kwargs) — returns dictionaries instead of objects. If you do not specify fields, it will return all fields as a dictionary.
values_list(*_fields, flat=False) — returns a list of tuples with the given fields. If only one field is passed and the parameter is passed flat=Truereturns a list with the values of this field for different objects.

# Создаем объекты
task = await Task.create(
    name="First task",
    description="First task description",
)

for i in range(0, 5):
    await Task.create(name="s", description="s"+str(i))

await Task.all().count()
# Output: 6

await Task.all().distinct().values('name', 'description')
# Output: [{'name': 'First task'}, {'name': 's'}]

await Task.filter(name="First task").exists()
# Output: True

from tortoise.functions import Count
await Task.all()\
		.group_by('name')\
		.annotate(name_count=Count('name'))\
		.values('name', 'name_count')
# Output: [{'name': 'First task', 'name_count': 1}, {'name': 's', 'name_count': 5}]

# Знак минус в order_by переворачивает выборку по полю.
await Task.all().order_by('-id')
# Output: [<Task: 6>, <Task: 5>, <Task: 4>, <Task: 3>, <Task: 2>, <Task: 1>]

await Task.all().limit(2)
# Output: [<Task: 1>, <Task: 2>]

await Task.all().offset(2)
# Output: [<Task: 3>, <Task: 4>, <Task: 5>, <Task: 6>]

await Task.all().values('name')
# Output: [{'name': 'First task'}, {'name': 's'}, {'name': 's'}, {'name': 's'}, {'name': 's'}, {'name': 's'}]

# Вернет плоский список из имен всех Task.
await Task.all().values_list('name', flat=True)
# Output: ['First task', 's', 's', 's', 's', 's']

Updating object fields:

There are two ways to update fields in objects.

Update via method update():

task = await Task.create(
	name="First task",
    description="First task description",
)

for i in range(0, 5):
    await Task.create(name="s", description="s"+str(i))

print(await Task.filter(name="s").values_list('description', flat=True))
# Output: ['s0', 's1', 's2', 's3', 's4']

await Task.filter(name="s")\
        .update(description="new description for all queryset")
    
print(await Task.filter(name="s").values_list('description', flat=True))
# Output: ['new description for all queryset', 'new description for all queryset', 'new description for all queryset', 'new description for all queryset', 'new description for all queryset']

Point change of an object through save():

task = await Task.create(
    name="First task",
    description="First task description",
)

for i in range(0, 5):
    await Task.create(name="s", description="s"+str(i))
    
task = await Task.get(id=1)
print(task.description)
# Output: First task description

task.description = "new description"
await task.save()

# Закрываем await вызов на Queryset в скобки
# чтобы получить объект вместо Queryset
print((await Task.get(id=1)).description)
# Output: new description

bulk_create and bulk_update:

There are also methods for bulk work with the database. bulk_create And bulk_update.

bulk_create will create the passed objects in one request. For example:

tasks_to_create = []
for i in range(0, 5):
    tasks_to_create.append(
        Task(
            name=f"Task name {i}",
            description=f"Task description {i}"
        )
    )
    
await Task.bulk_create(tasks_to_create)

print(await Task.all())
# Output: [<Task: 1>, <Task: 2>, <Task: 3>, <Task: 4>, <Task: 5>]

bulk_update will replace specific fields on the passed objects:

tasks_to_create = []
for i in range(0, 5):
    tasks_to_create.append(
        Task(
            name=f"Task name {i}",
            description=f"Task initial description {i}"
        )
    )
    
await Task.bulk_create(tasks_to_create)

print(await Task.all().values_list('description', flat=True))
# Output: ['Task initial description 0', 'Task initial description 1', 'Task initial description 2', 'Task initial description 3', 'Task initial description 4']

tasks = await Task.all()
for i, task in enumerate(tasks):
    task.description = f"New task description {i}"

await Task.bulk_update(tasks, ['description'])

print(await Task.all().values_list('description', flat=True))
# Output: ['New task description 0', 'New task description 1', 'New task description 2', 'New task description 3', 'New task description 4']

Removing an object:
If we already have an object:

task = await Task.get(id=1)
await task.delete()

You don't have to declare the object, but can do it like this:

await Task.get(id=1).delete()

It also works with a list of objects:

# Через объект
tasks = await Task.all()
print(tasks)
# Output: [<Task: 1>, <Task: 2>, <Task: 3>]

await Task.all().delete()
# или
# await tasks.delete()

tasks = await Task.all()
print(tasks)
# Output: []

Working with DatetimeField

Tortoise ORM allows you to make convenient selections by Datetime fields. Doesn't work with Sqlite3but works with PostgresSQL and MySQL.

Documentation

We will use the model from the previous chapter:

class Task(models.Model):
    id = fields.IntField(primary_key=True)
    name = fields.CharField(max_length=256)
    description = fields.CharField(max_length=500)
    date_created = fields.DatetimeField(auto_now_add=True)
    date_updated = fields.DatetimeField(auto_now=True)

    class Meta:
        table = "tasks"

# Возможные фильтры для выборки из документации
class DatePart(Enum):
	year = "YEAR"
	quarter = "QUARTER"
	month = "MONTH"
	week = "WEEK"
	day = "DAY"
	hour = "HOUR"
	minute = "MINUTE"
	second = "SECOND"
	microsecond = "MICROSECOND"

# Примеры:
await Task.filter(date_created__year=2024)
await Task.filter(date_created__month=6)
await Task.filter(date_created__day=10)

Working with JSONField

Under the hood, when a JSONField is specified, Tortoise ORM creates a JSONB field in the database table.

In PostgreSQL and MySQL you can use contains, contained_by And filter options in JSONField.

Examples from the documentation:

class JSONModel:
    data = fields.JSONField()

await JSONModel.create(data=["text", 3, {"msg": "msg2"}])
obj = await JSONModel.filter(data__contains=[{"msg": "msg2"}]).first()

await JSONModel.create(data=["text"])
await JSONModel.create(data=["tortoise", "msg"])
await JSONModel.create(data=["tortoise"])

objects = await JSONModel.filter(data__contained_by=["text", "tortoise", "msg"])

class JSONModel:
    data = fields.JSONField()

await JSONModel.create(data={"breed": "labrador",
                             "owner": {
                                 "name": "Boby",
                                 "last": None,
                                 "other_pets": [
                                     {
                                         "name": "Fishy",
                                     }
                                 ],
                             },
                         })

obj1 = await JSONModel.filter(data__filter={"breed": "labrador"}).first()
obj2 = await JSONModel.filter(data__filter={"owner__name": "Boby"}).first()
obj3 = await JSONModel.filter(data__filter={"owner__other_pets__0__name": "Fishy"}).first()
obj4 = await JSONModel.filter(data__filter={"breed__not": "a"}).first()
obj5 = await JSONModel.filter(data__filter={"owner__name__isnull": True}).first()
obj6 = await JSONModel.filter(data__filter={"owner__last__not_isnull": False}).first()

There is no function in the standard Tortoise ORM jsonb_sethowever in issue The project has an implementation from one of the contributors:

from tortoise.expressions import Fz
from pypika.terms import Function

class JsonbSet(Function):
    def __init__(self, field: F, path: str, value: Any, create_if_missing: bool = False):
        super().__init__("jsonb_set", field, path, value, create_if_missing)

json = await JSONModel.create(data={"a": 1})
json.data_default = JsonbSet(F("data"), "{a}", '3') # in fact '3' is integer 
await json.save()

Description of connections

Tortoise ORM supports basic One-to-One, One-to-Many, Many-to-Many relationships. Unfortunately, Tortoise ORM cannot work with Polymorphic relations.

One-to-One:
Models:

class Task(models.Model):
    id = fields.IntField(primary_key=True)
    name = fields.CharField(max_length=256)
    description = fields.CharField(max_length=500)
    date_created = fields.DatetimeField(auto_now_add=True)
    date_updated = fields.DatetimeField(auto_now=True)

	# Выставляем поле для type hinting
    additional_info: fields.OneToOneRelation["AdditionalTaskInfo"]

    class Meta:
        table = "tasks"


class AdditionalTaskInfo(models.Model):
    id = fields.IntField(primary_key=True)
    additional_name = fields.CharField(max_length=256)
    additional_description = fields.CharField(max_length=500)
    
    task = fields.OneToOneField(
        model_name="models.Task",
        related_name="additional_info",
        on_delete=fields.CASCADE,
    )

    class Meta:
        table = "additionaltaskinfos"

task = await Task.create(
    name="First task",
	description="First task description",
)

print(await task.additional_info)
# Output: None

await AdditionalTaskInfo.create(
    task=task,
    additional_name="add name",
    additional_description="add desc"
)

print(await task.additional_info)
# Output: <AdditionalTaskInfo>

await AdditionalTaskInfo.create(
    task=task,
    additional_name="add name",
    additional_description="add desc"
)
# Exception: tortoise.exceptions.IntegrityError: UNIQUE constraint failed: additionaltaskinfos.task_id

One-to-Many:
Models:

class Task(models.Model):
    id = fields.IntField(primary_key=True)
    name = fields.CharField(max_length=256)
    description = fields.CharField(max_length=500)
    date_created = fields.DatetimeField(auto_now_add=True)
    date_updated = fields.DatetimeField(auto_now=True)

	# Type hinting
    reminders: fields.ForeignKeyRelation["TaskReminder"]

    class Meta:
        table = "tasks"


class TaskReminder(models.Model):
    id = fields.IntField(primary_key=True)
    remind_date = fields.DatetimeField()
    status = fields.IntField(default=0)
    
    task = fields.ForeignKeyField(
        model_name="models.Task",
        related_name="reminders",
        on_delete=fields.CASCADE,
    )

    class Meta:
        table = "task_reminders"

Example of work:

task = await Task.create(
    name="First task",
    description="First task description",
)
    
print(await task.reminders)
# Output: []

# Создаем 5 напоминалок через объект task
for i in range(0, 5):
    await TaskReminder.create(
        remind_date=datetime.now() + timedelta(hours=i),
        task=task,
    )

# Также можно создавать по pk 
await TaskReminder.create(
    remind_date=datetime.now() + timedelta(hours=i),
    task_id=task.pk,
    status=1,
)

# Необходимо await на связи, чтобы получить модели
print(await task.reminders)
# Output: [<TaskReminder: 1>, <TaskReminder: 2>, <TaskReminder: 3>, <TaskReminder: 4>, <TaskReminder: 5>, <TaskReminder: 6>]

# Также поле связи работает как менеджер, в котором мы можем фильтровать запрос
print(await task.reminders.filter(status=1))
# Output: [<TaskReminder: 6>]

task2 = await Task.get(id=1).prefetch_related("reminders")

# Можно запрефетчить объекты, чтобы одним вызовом получить все напоминалки
reminders = []
for reminder in task2.reminders:
    reminders.append(reminder)
print(reminders)
# Output: [<TaskReminder: 1>, <TaskReminder: 2>, <TaskReminder: 3>, <TaskReminder: 4>, <TaskReminder: 5>, <TaskReminder: 6>]

task3 = await Task.get(id=1)

# Также все объекты можно получить через асинхронный for
async_reminders = []
async for reminder in task3.reminders:
    async_reminders.append(reminder)
print(reminders)
# Output: [<TaskReminder: 1>, <TaskReminder: 2>, <TaskReminder: 3>, <TaskReminder: 4>, <TaskReminder: 5>, <TaskReminder: 6>]

Many-to-Many:
Models:

class Course(models.Model):
    id = fields.IntField(primary_key=True)
    name = fields.CharField(max_length=256)
    
    students: fields.ManyToManyRelation["Student"]


class Student(models.Model):
    id = fields.IntField(primary_key=True)
    name = fields.CharField(max_length=256)

    courses = fields.ManyToManyField(
        model_name="models.Course",
        through="student_courses",
        related_name="students",
    )

Example of work:

student = await Student.create(name="Student")
course = await Course.create(name="Course")
course2 = await Course.create(name="Course 2")

await student.courses.add(course)
await student.courses.add(course2)

print(await student.courses)
# Output: [<Course: 1>, <Course: 2>]

print(await course.students)
# Output: [<Student: 1>]

await student.courses.remove(course)
await course2.students.remove(student)

print(await student.courses)
# Output: []

print(await course.students)
# Output: []

Working with M2M associations

In Tortoise ORM we can manually create associations through an intermediate model. Working with them will not be so convenient and additional challenges will appear. But still – at least it is possible.

Models:

class Student(models.Model):
    id = fields.IntField(primary_key=True)
    name = fields.CharField(max_length=256)
    scm2m: fields.ReverseRelation["StudentCourseM2M"]


class Course(models.Model):
    id = fields.IntField(primary_key=True)
    name = fields.CharField(max_length=256)
    scm2m: fields.ReverseRelation["StudentCourseM2M"]


class StudentCourseM2M(models.Model):
    id = fields.IntField(primary_key=True)

    due_date = fields.DatetimeField()

    student = fields.ForeignKeyField(
        model_name="models.Student",
        related_name="scm2m",
        on_delete=fields.CASCADE,
    )
    course = fields.ForeignKeyField(
        model_name="models.Course",
        related_name="scm2m",
        on_delete=fields.CASCADE,
    )

Examples:

student = await Student.create(name="student")
course = await Course.create(name="course")

scm2m = await StudentCourseM2M.create(
    due_date=datetime.now() + timedelta(days=1),
    student=student,
    course=course,
)

print(scm2m.student)
# Output: <Student>
print(scm2m.course)
# Output: <Course>

# Фетчим связанные с м2м моделью поля
await student.fetch_related("scm2m__course")
await course.fetch_related("scm2m__student")
    
print(student.scm2m[0].due_date)
# Output: 2024-06-16 19:28:16.599730+00:00
print(course.scm2m[0].due_date)
# Output: 2024-06-16 19:28:16.599730+00:00
print(student.scm2m[0].course.name)
# Output: course
print(course.scm2m[0].student.name)
# Output: student

Working with transactions

Tortoise ORM has two ways to work with transactions: decorator and context manager.
Documentation

Model:

class Task(models.Model):
    id = fields.IntField(primary_key=True)
    name = fields.CharField(max_length=256)
    description = fields.CharField(max_length=500)
    date_created = fields.DatetimeField(auto_now_add=True)
    date_updated = fields.DatetimeField(auto_now=True)

    class Meta:
        table = "tasks"

Via decorator:

from tortoise.transactions import atomic

@atomic
async def change_in_transaction():
    tasks = await Task.all()
    
    counter = 0
    for i, task in enumerate(tasks):
        if counter >= 3:
            raise Exception("Something went wrong")

        task.description = f'new task description {i}'
        await task.save()
        counter += 1

task = await Task.create(
    name="First task",
    description="First task description",
)

for i in range(0, 5):
    await Task.create(name="s", description="s"+str(i))

print(await Task.all().values_list('description', flat=True))
# Output: ['First task description', 's0', 's1', 's2', 's3', 's4']
    
try:
    await change_in_transaction()
except Exception:
    print("Something went wrong inside transation")
    # Output: Something went wrong inside transation

print(await Task.all().values_list('description', flat=True))
# Output: ['First task description', 's0', 's1', 's2', 's3', 's4']
# Данные остались неизменными несмотря на то, что мы сохранили
# изменения описания в нескольких задачах

Via context manager:

from tortoise.transactions import in_transaction

task = await Task.create(
    name="First task",
    description="First task description",
)

for i in range(0, 5):
    await Task.create(name="s", description="s"+str(i))

print(await Task.all().values_list('description', flat=True))
# Output: ['First task description', 's0', 's1', 's2', 's3', 's4']

try:
    async with in_transaction():
        tasks = await Task.all()
    
        counter = 0
        for i, task in enumerate(tasks):
            if counter >= 3:
                raise Exception("Something went wrong")

            task.description = f'new task description {i}'
            await task.save()
            counter += 1
except Exception:
    print("Something went wrong inside the transaction")
    # Output: Something went wrong inside the transaction

print(await Task.all().values_list('description', flat=True))
# Output: ['First task description', 's0', 's1', 's2', 's3', 's4']
# Данные также остались неизменными

Indexes

Documentation
By default, Tortoise ORM uses BTree for indexing if the field is passed the parameter db_index=True or indexes are set to class Meta. But if another type of indexing is needed, such as, FullTextIndex in MySQL, you need to use tortoise.indexes.Index or his heirs.

Example with indexes for MySQL:

from tortoise import Model, fields
from tortoise.contrib.mysql.fields import GeometryField
from tortoise.contrib.mysql.indexes import FullTextIndex, SpatialIndex


class Index(Model):
    full_text = fields.TextField()
    geometry = GeometryField()

    class Meta:
        indexes = [
            FullTextIndex(fields={"full_text"}, parser_name="ngram"),
            SpatialIndex(fields={"geometry"}),
        ]

Ready-made indexes can be found in modules tortoise.contrib.mysql.indexes And tortoise.contrib.postgres.indexes

It can also be expanded tortoise.indexes.Index and enter your index.
Example from the documentation:

from typing import Optional, Set
from pypika.terms import Term
from tortoise.indexes import Index

class FullTextIndex(Index):
    INDEX_TYPE = "FULLTEXT"

    def __init__(
        self,
        *expressions: Term,
        fields: Optional[Set[str]] = None,
        name: Optional[str] = None,
        parser_name: Optional[str] = None,
    ):
        super().__init__(*expressions, fields=fields, name=name)
        if parser_name:
            self.extra = f" WITH PARSER {parser_name}"

However, for Postgres the extension is different, it is necessary to inherit from tortoise.contrib.postgres.indexes.PostgresSQLIndex:

class BloomIndex(PostgreSQLIndex):
    INDEX_TYPE = "BLOOM"

Functions, Aggregates and Expressions

Functions, aggregates and expressions allow you to create complex SQL queries to retrieve/process data in a specific form. Tortoise ORM can work with all of them.

More details in documentation of functions/units And expression documentation

Tortoise ORM supports the following list of features (tortoise.functions.*):

Trim
Length
Coalesce
Lower
Upper
Concat

Database-specific functions for randomness:

tortoise.contrib.mysql.functions.Rand
tortoise.contrib.postgres.functions.Random
tortoise.contrib.sqlite.functions.Random

Tortoise ORM supports the following list of aggregates (tortoise.functions):

Tortoise ORM also allows you to extend functions. Using the JSON_SET function as an example:

from tortoise.expressions import F
from pypika.terms import Function

class JsonSet(Function):
    def __init__(self, field: F, expression: str, value: Any):
        super().__init__("JSON_SET", field, expression, value)

json = await JSONFields.create(data_default={"a": 1})
json.data_default = JsonSet(F("data_default"), "$.a", 2)
await json.save()

# or use queryset.update()
sql = JSONFields.filter(pk=json.pk).update(data_default=JsonSet(F("data_default"), "$.a", 3)).sql()
print(sql)
# UPDATE jsonfields SET data_default=JSON_SET(`data_default`,'$.a',3) where id=1

Tortoise ORM supports the following list of expressions (tortoise.expressions):

Timezones

Working with time zones is inspired by Django ORM, but has some differences. There are two parameters in the config use_tz And timezonewhich are exhibited in Tortoise.init. Also, depending on the database, the behavior may differ.

When use_tz=TrueTortoise ORM will always store UTC time in the database regardless of the time zone.
MySQL uses the field DATETIME(6)
PostgreSQL uses TIMESTAMPZ
SQLite uses TIMESTAMP when generating schematics.

For TimeField MySQL uses TIME(6)PostgreSQL uses TIMETZSQLite uses TIME.

Parameter timezone determines which time zone is used when selecting from the database, regardless of which time zone is set in the database itself. It is also necessary to use tortoise.timezone.now() instead of datetime.datetime.now().

Signals

Tortoise ORM supports signals. There are currently four signals:

pre_save
post_save
pre_delete
post_delete

The signals are implemented as a decorator. You can find them in the module toroise.signals.*.
Working with signals looks like this:

class Task(models.Model):
    id = fields.IntField(primary_key=True)
    name = fields.CharField(max_length=256)
    description = fields.CharField(max_length=500)
    date_created = fields.DatetimeField(auto_now_add=True)
    date_updated = fields.DatetimeField(auto_now=True)

    class Meta:
        table = "tasks"

from tortoise.signals import pre_save
@pre_save(Task)
async def task_pre_save(model, instance, db_client, *args):
    instance.description = instance.description + " addition"

task = await Task.create(
    name="First task",
    description="First task description",
)
print(task.description)
# Output: First task description addition

Connection pool and database connection configuration

In Tortoise ORM, you can set the connection pool size (min_size, max_size). Depending on the DB, the configuration parameters may differ.

For more information on how to configure your database, please see documentation

In the basic example we use the connection via db_url. But you can use the config parameter to pass the necessary data and fine-tune the connection configuration.

The average config will look something like this:

await Tortoise.init(
    config={
        "connections": {
            "default": {
                "engine": "tortoise.backends.asyncpg",
                "credentials": {
                    "database": None,
                    "host": "127.0.0.1",
                    "password": "moo",
                    "port": 54321,
                    "user": "postgres",
                    "minsize": 1,
                    "maxsize": 10,
                }
            }
        },
        "apps": {
            "models": {
                "models": ["some.models"],
                "default_connection": "default",
            }
        },
    }
)

Logging

Tortoise ORM currently has two loggers: tortoise.db_client And tortoise tortoise.db_clientлоггирует информацию об исполнении запросов, аtortoise` logs runtime information.

If you want to control TortoiseORM log behavior, such as SQL debugging, you can configure the logs yourself:

import logging

fmt = logging.Formatter(
    fmt="%(asctime)s - %(name)s:%(lineno)d - %(levelname)s - %(message)s",
    datefmt="%Y-%m-%d %H:%M:%S",
)
sh = logging.StreamHandler(sys.stdout)
sh.setLevel(logging.DEBUG)
sh.setFormatter(fmt)

# will print debug sql
logger_db_client = logging.getLogger("tortoise.db_client")
logger_db_client.setLevel(logging.DEBUG)
logger_db_client.addHandler(sh)

logger_tortoise = logging.getLogger("tortoise")
logger_tortoise.setLevel(logging.DEBUG)
logger_tortoise.addHandler(sh)

Also, you can add colors to the logger via pygments, more details in documentation

Router for several databases

The easiest way to use multiple databases is to create a database routing scheme. The default routing scheme will try to connect objects to the original database (i.e. if an object was requested from the foo database, then storing the object will also be in the foo database). The default routing scheme will always use the default database if no other is passed.

Setting up a custom router is very simple: you just need to create a class that will implement the methods db_for_read(self, model) And db_for_write(self, model):

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

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

The methods return the names of the databases that were specified in the config.

Accordingly, the config should look something like this:

config = {
    "connections": {"master": "sqlite:///tmp/test.db", "slave": "sqlite:///tmp/test.db"},
    "apps": {
        "models": {
            "models": ["__main__"],
            "default_connection": "master",
        }
    },
    "routers": ["path.Router"],
    "use_tz": False,
    "timezone": "UTC",
}
await Tortoise.init(config=config)
# или
routers = config.pop('routers')
await Tortoise.init(config=config, routers=routers)

Page in documentation

Overriding Object Manager

In Tortoise ORM, as in Django ORM, you can override the manager of the model. The manager is an interface through which queries are created for Tortoise ORM models.

There are two ways to work with managers: you can override the standard manager or add an additional one.

For example:

from tortoise.manager import Manager

class StatusManager(Manager):
    def get_queryset(self):
        return super(StatusManager, self).get_queryset().filter(status=1)


class ManagerModel(Model):
    status = fields.IntField(default=0)
    # Добавление еще одного менеджера
    # В этом случае, мы сохраняем стандартный менеджер
    # в поле all_objects.
    all_objects = Manager()

	# Переопределение стандартного менеджера
    class Meta:
        manager = StatusManager()

In the example above, the queries are, for example, get or filterwill not be able to return objects from status=0To achieve this, we can use the standard manager in all_objects:

m1 = await ManagerModel.create()
m2 = await ManagerModel.create(status=1)

assert await ManagerModel.all().count() == 1
assert await ManagerModel.all_objects.all() == 2

Description of field extension

In Tortoise ORM, you can extend field types. To do this, you need to inherit from a field of a certain type that can be represented in the DB.

IN documentation An example of extending a CharField field to work with Enum is presented.

from enum import Enum
from typing import Type

from tortoise import ConfigurationError
from tortoise.fields import CharField


class EnumField(CharField):
    """
    Пример расширения CharField который сериализует Enum в и из str
    в представление в БД
    """

    def __init__(self, enum_type: Type[Enum], **kwargs):
        super().__init__(128, **kwargs)
        if not issubclass(enum_type, Enum):
            raise ConfigurationError("{} is not a subclass of Enum!".format(enum_type))
        self._enum_type = enum_type

    def to_db_value(self, value: Enum, instance) -> str:
        return value.value

    def to_python_value(self, value: str) -> Enum:
        try:
            return self._enum_type(value)
        except Exception:
            raise ValueError(
                "Database value {} does not exist on Enum {}.".format(value, self._enum_type)
            )

When you inherit, make sure that to_db_value returns the same type as the parent class (in this case str) So what to_python_value takes the same type as the parameter value (in this case str)

I can also give an example of expanding the field for the type tstzrange from PostgreSQL:

from asyncpg import Range
from tortoise import fields

class DateTimeRangeField(fields.Field, Range):
    SQL_TYPE = "tstzrange"

    def to_python_value(self, value: Any) -> Any:
        return value

A little bit about Aerich (library for working with migrations for Tortoise ORM)

Every self-respecting ORM should have a tool for working with migrations, and Aerich for Tortoise ORM is just that.

Aerich is almost no different from most solutions in its field of activity. Through the CLI, Aerich allows you to set up a project for migrations, initialize a database, create models based on tables in an already created database, create migrations and roll them out. Also, if necessary, roll back certain migrations. Migrations are created in pure SQL.

Aerich can also be used directly from Python code if you want to manage migrations via Python scripts.

I won't repeat the documentation for the tool here, if you're interested, I'll leave a link to documentation for more complete information.

FastAPI + Tortoise ORM Template

To make your introduction to the stack more enjoyable and easier, I have made a template for your stack applications and tests.

Github repository

The repository is prepared for CookieCutter. To build a template for your project, install Cookie Cutter and run the command:

cookiecutter git@github.com:SquonSerq/fastapi-tortoise-template.git

It is not debugged, but should start out of the box. The project has a simple structure for the project, uses Poetry to work with dependencies and has support for containerization via Docker.

The template will be updated, I will be glad to receive your suggestions/comments in issues.
In the future, I plan to add work with tests via pytest.

Briefly about FastAPI Admin

There is one interesting library on the Internet: FastAPI Admin

It was created by developer long2ice, one of the FastAPI contributors. For me, FastAPI Admin was one of the reasons to switch from SQLAlchemy to Tortoise.
However, when I tried to introduce it into my project, I couldn’t do it.

I don't remember exactly, but I decided that it was taking me too much time and I had to throw out the idea with FastAPI Admin. It was installed, but it looked and worked extremely clumsily.

Perhaps, a year after my last attempt, they rolled out updates that will make it easier to deploy. If someone managed to deploy this miracle on their project, I will be very glad to hear about the experience in the comments.

Conclusion

Thank you for reading the material to the end. I hope it was useful. I really wanted to collect something intelligible on Tortoise ORM, since there are practically no sources in Russian. Share your experience of working with this library in the comments. I will be glad to chat and listen to the opinions of Habrausers.

I have collected in one article everything you need to know about the tool and my experience working with it

How are things with FastAPI and SQLAlchemy?

Description of Tortoise ORM

Comparing Tortoise ORM Performance with Other ORM Libraries

Basic Tortoise ORM Setup

Queryset Evaluation in Tortoise ORM

Description of possible field types

Description of simple operations

Working with DatetimeField

Working with JSONField

Description of connections

Working with M2M associations

Working with transactions

Indexes

Functions, Aggregates and Expressions

Timezones

Signals

Connection pool and database connection configuration

Logging

Router for several databases

Overriding Object Manager

Description of field extension

A little bit about Aerich (library for working with migrations for Tortoise ORM)

FastAPI + Tortoise ORM Template

Briefly about FastAPI Admin

Conclusion

How a front-end developer passed the PostgresPro certification

The History of AI Self-Creation

what about patents?

An in-depth analysis of testing widgets in Flutter. Part II. Finder and WidgetTester Classes

Automount SMB Shares Using KRB5

The most successful and most scandalous Data Science project: Cambridge Analytica

Leave a Reply Cancel reply

How are things with FastAPI and SQLAlchemy?

Description of Tortoise ORM

Comparing Tortoise ORM Performance with Other ORM Libraries

Basic Tortoise ORM Setup

Queryset Evaluation in Tortoise ORM

Description of possible field types

Description of simple operations

Working with DatetimeField

Working with JSONField

Description of connections

Working with M2M associations

Working with transactions

Indexes

Functions, Aggregates and Expressions

Timezones

Signals

Connection pool and database connection configuration

Logging

Router for several databases

Overriding Object Manager

Description of field extension

A little bit about Aerich (library for working with migrations for Tortoise ORM)

FastAPI + Tortoise ORM Template

Briefly about FastAPI Admin

Conclusion

Similar Posts

Leave a Reply Cancel reply