Привет, Хабр! Меня зовут Даниил Лихачев, я Python backend developer в диджитал-продакшене Далее. Сегодня я хотел бы представить вашему вниманию асинхронную библиотеку для работы с базами данных под названием Tortoise ORM. Это обзорная статья, чтобы показать, что из себя представляет данная библиотека и для каких проектов она подойдет. Также на основе своего опыта постараюсь осветить аспекты, в которых Tortoise ORM хороша и удобна, а также те, в которых ее возможностей может не хватать и как это обойти. Также бонусом предоставлю свой шаблон в стеке FastAPI + Tortoise ORM.

Содержание статьи

• Как там у FastAPI с SQLAlchemy?

• Описание Tortoise ORM

• Сравнение производительности Tortoise ORM с другими ORM-библиотеками

• Сетап Tortoise ORM (базово)

• Queryset Evaluation в Tortoise ORM

• Описание возможных типов полей

• Описание простых операций

• Работа с Datetime полями

• Работа с JSONB полями

• Описание связей

• Работа с M2M ассоциациями

• Работа с транзакциями

• Индексы

• Функции, агрегаты и выражения

• Таймзоны

• Сигналы

• Логгирование

• Роутер для нескольких БД

• Переопределение Object Manager

• Описание расширений полей

• Connection pool и конфигурация подключения к БД

• Немного про Aerich (библиотека для миграций в связке с Tortoise ORM)

• Шаблон для FastAPI + Tortoise ORM

• Коротко о FastAPI Admin

• Заключение

Как там у FastAPI с SQLAlchemy?

SQLAlchemy (или Алхимия) на данный момент является де-факто orm в стеке с FastAPI. Уверен, есть много почитателей такого стека, которые получают удовольствие от работы со связкой этих инструментов.

Относительно недавно разработчик FastAPI Tiangolo начал активно продвигать SQLModel в работу с FastAPI, с помощью которой можно делать как Pydantic классы, так и сразу модели SQLAlchemy. Это снижает количество головной боли, однако многие вещи в работе с SQLAlchemy все еще остаются неудобными. Не могу сказать, что у меня большой опыт в Алхимии. Что-то из этого может быть некорректным, так что прошу иметь это в виду. Вот негативные пункты в работе с этим инструментом, которые я подметил для себя:

Прокидывание db_session:

Для того, чтобы работать с SQLAlchemy, приходится прокидывать db_session от контроллера вниз по слоям абстракции, что крайне неудобно. Например, у нас есть 4 слоя абстракции, на нижнем из которых производится запрос к БД через Алхимию. В этот момент необходимо прокинуть от !!контроллера!! до этого слоя db_session, что крайне неудобно. Это постоянный аргумент для передачи, о котором нужно помнить.

Не очень удобное составление запросов:

Лично мне не нравится стиль написания запросов в SQLAlchemy. Приходится писать много ненужного кода, чтобы составить простецкий запрос. Да, повышается уровень контроля над ними, но даже самые простые могут занимать по 3-4 строчки. Допустим, в сервисе этих запросов куча — тогда код раздувается как на дрожжах.

Не очень удобная документация:

Скорее всего, проблема во мне, но я просто теряюсь в документации SQLAlchemy. Skill issue, скажите вы, и, возможно, будете правы.

Составление дополнительного слоя для CRUD-подобных вещей:

Кроме бизнес-логики, приходится руками прописывать функции/методы для работы с моделями, например: получение по pk, добавление объекта, удаление и т.д.
Раньше в темплейте от разработчиков FastAPI был CRUD, который можно навесить на любую модель и базовые операции начинали работать. Однако даже с таким подходом все равно приходится писать какие-то кастомные выборки в CRUD конкретной модели. Больше boilerplate кода богу boilerplate кода!

Эти пункты появились напрямую от моего опыта использования SQLAlchemy. Все фломастеры на вкус разные, так что вы можете быть со мной не согласны и написать об этом в комментариях. Мне эти пункты мешали при работе с Алхимией. Чтобы избавиться от этого инструмента, я стал искать замену, которой стал TortoiseORM. Мы взяли на ее проект для пробы, и в итоге на ней и остановились.

Описание Tortoise ORM

Tortoise ORM — асинхронная библиотека для работы с базой данных. Первый коммит был залит на Github 29 марта 2018 года. На момент написания статьи, у Tortoise ORM 4.3к звезд на Github и вышла 0.21.3 версия, на базе которой и будем рассматривать библиотеку в статье. Под капотом TortoiseORM использует pypika для составления SQL Query.

Tortoise ORM будет крайне знакомой, можно даже сказать родной всем, кто до этого работал с Django ORM. Дело в том, что разработчики создали свою ORM, вдохновляясь Django ORM. Однако Django ORM идет только в пачке с Django и, хотя в Django добавляются возможности для асинхронной работы, это все еще остается не тем инструментом, который ты выберешь, думая об асинхронности.

Разработчики указывают, что Tortoise ORM — все еще молодая библиотека, и могут появляться изменения, ломающие существующий код при обновлении. Стоит иметь это в виду при использовании ее в своих проектах.

Сравнение производительности Tortoise ORM с другими ORM-библиотеками

Разработчики Tortoise ORM произвели сравнения производительности разных ORM-библиотек для Python. Как можно заметить из их бенчмарка, Tortoise ORM хорошо конкурирует с другими ORM-библиотеками, часто меняясь местами с Pony ORM. Разработчики нацелены на производительность, и этот бенчмарк помог создателям найти места, которые замедляют работу библиотеки.

Встают, конечно, вопросы по части честности таких тестов, так как необходимо учитывать много переменных для объективного сравнения.

Более подробно с бенчмарком можно ознакомиться на Github

Базовая настройка Tortoise ORM

Для начала, необходимо поставить нужные пакеты:
pip install tortoise-orm

Дополнительно можно поставить драйвера под вашу БД:

PostgreSQL:
pip install tortoise-orm[asyncpg]

MySQL:
pip install tortoise-orm[asyncmy]

Microsoft SQL Server / Oracle (протестированы не полностью):
pip install tortoise-orm[asyncodbc]

Дальше задаем функцию запуска ОРМ, которую затем вызываем на старте приложения:

from tortoise import Tortoise

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

После этого вызываем init() на старте нашего приложения — и можно использовать модели в асинхронных функциях.

Важно помнить о том, что необходимо закрывать коннект к базе.
Для этого используем await Tortoise.close_connection()

В дальнейшем, будем использовать подобный набросок кода для рассмотрения возможностей 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 в Tortoise ORM

Если вы работали с Django ORM, то знаете о Queryset Evaluation. Это момент, когда ваш запрос преобразуется в конкретные объекты. В Django это сделано неявно, поэтому очень важно знать, когда ваш Queryset превратится в объекты.

В Tortoise ORM с этим намного проще, так как асинхронность позволила сделать Queryset Evaluation явным. Queryset строится ровно до тех пор, пока на нем не будет вызван await.

Например:

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

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

Описание возможных типов полей

Tortoise ORM поддерживает следующие типы полей (tortoise.fields):

• BigIntField

• BinaryField

• BooleanField

• CharEnumField

• CharField

• DateField

• DatetimeField

• TimeField

• DecimalField

• FloatField

• IntEnumField

• IntField

• JSONField

• SmallIntField

• TextField

• TimeDeltaField

• UUIDField

Также есть поля, в зависимости от используемой БД.
MySQL (tortoise.contrib.mysql.fields):

• GeometryField

• UUIDField

PostgreSQL (tortoise.contrib.postgres.fields):

• TSVectorField

Также можно расширять поля, добавляя собственный функционал. Об этом расскажу дальше в статье.

Больше о полях в документации: ссылка

Описание простых операций

У Tortoise ORM очень простой API, благодаря которому работать с ней одно удовольствие. Опять-таки, всем разработчикам, кто работал с Django ORM, все описанное ниже будет до боли знакомо.

Коротко пройдемся по API, которое предоставляет Tortoise ORM.
Более подробно в документации: ссылка

Для примера будем использовать следующую модель:

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"

Создание объекта:

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

Однако, нет смысла каждый раз создавать объекты таким образом, потому что разработчики сделали алиас для создания, который делает то же самое, но быстрее:

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

Получение объекта по полю:

task = await Task.get(id=1)

Если мы вызовем код сверху, но таски с таким id не будет, нам вывалится исключение, ровно как и в Django.

В Django такой кейс часто обходится двумя конструкциями:

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

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

В Tortoise ORM хоть и можно, но нет необходимости в таких конструкциях, т.к. есть метод get_or_none(**fields)

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

Также можно получать объекты по нескольким полям:

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

Через метод get_or_create(**fields) можно либо получить объект с заданными полями, либо создать новый, если ни один существующий объект не попал под условия:

# Возвращает 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

Получение списка объектов:
Точно так же, как в Django, можно получать список объектов через метод filter():

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

Tortoise ORM включает в себя следующие методы для выборки:

• filter(*args, **kwargs) — фильтр по заданным значениям полей

• exclude(*args, **kwargs) — фильтр, исключающий объекты, попадающие под условия выборки

• all() — получение всех объектов модели

• first() — ограничивает выборку до одного объекта и возвращает его, вместо списка

• annotate() — аннотация данных в выборку

Как и в Django, в методах фильтрации можно указывать дополнительные фильтры по полям. Например, следующий код выдаст нам список объектов, названия которых включены в переданный список:

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

Tortoise ORM поддерживает следующий фильтры полей:

• not

• in — проверяет попадает ли значение поля в переданный список

• not_in

• gte — больше или равно переданному значению

• gt — больше чем переданное значение

• lte — меньше или равно переданному значению

• lt — меньше чем переданное значение

• range — между переданными двумя значениями

• isnull — является ли поле null

• not_isnull — не является ли поле null

• contains — поле содержит подстроку

• icontains — contains вне зависимости от регистра

• startswith — поле начинается со значения

• istartswith — startswith вне зависимости от регистра

• endswith — поле заканчивается значением

• iendswith — endswith вне зависимости от регистра

• iexact — сравнение без учета регистра

• search — полнотекстовый поиск

Примеры:

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(t)
# 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>]

Также есть методы, преобразующие выборку в определенный вид:

• count() — возвращает количество объектов, вместо самих объектов.

• distinct() — возвращает уникальные значения из выборки. Имеет смысл только в комбинации с values() и values_list(), т.к. фильтрует пришедшие поля.

• exists() — возвращает True/False на предмет наличия объектов в выборке.

• group_by() — возвращает список словарей или кортежей с group by. Должен быть вызван перед values() или values_list().

• order_by() — сортирует выборку по полю/полям.

• limit() — ограничивает выборку на заданную длину.

• offset() — оффсет для объектов выборки из таблицы.

• values(*args, **kwargs) — возвращает словари вместо объектов. Если не задать поля, то вернет все поля в виде словаря.

• values_list(*_fields, flat=False) — возвращает список кортежей с заданными полями. Если передано только одно поле и передан параметр flat=True, возвращает список со значениями этого поля у разных объектов.

# Создаем объекты
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']

Обновление полей объекта:

Есть два способа обновления полей в объектах.

Обновление через метод 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']

Точечное изменение объекта через 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 и bulk_update:

Также есть методы для bulk работы с БД. bulk_create и bulk_update.

bulk_create создаст переданные объекты одним запросом. Например:

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 заменит конкретные поля у переданных объектов:

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']

Удаление объекта:
Если у нас уже есть объект:

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

Можно не объявлять объект, а сделать следующим образом:

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

Так же работает и со списком объектов:

# Через объект
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: []

Работа с DatetimeField

Tortoise ORM позволяет делать удобные выборки по Datetime полям. Не работает с Sqlite3, но работает с PostgresSQL и MySQL.

Документация

Будем использовать модель из прошлой главы:

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)

Работа с JSONField

Под капотом, когда задается JSONField, Tortoise ORM создает JSONB поле в таблице БД.

В PostgreSQL и MySQL можно использовать contains, contained_by и filter опции в JSONField.

Примеры из документации:

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

В стандартной Tortoise ORM нет функции jsonb_set, однако в issue проекта есть реализация от одного из контрибьюторов:

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

Описание связей

Tortoise ORM поддерживает базовые One-to-One, One-to-Many, Many-to-Many связи. К сожалению, Tortoise ORM не умеет работать с полиморфными связями (Polymorphic relations).

One-to-One:
Модели:

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:
Модели:

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"

Пример работы:

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:
Модели:

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

Пример работы:

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: []

Работа с M2M-ассоциациями

В Tortoise ORM мы можем руками создавать ассоциации через промежуточную модель. Работа с ними будет не такая уж и удобная и появятся дополнительные вызовы. Но все же — она хотя бы возможна.

Модели:

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,
    )

Примеры:

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

Работа с транзакциями

У Tortoise ORM есть два способа работы с транзакциями: декоратор и контекстный менеджер.
Документация

Модель:

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.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']
# Данные остались неизменными несмотря на то, что мы сохранили
# изменения описания в нескольких задачах

Через контекстный менеджер:

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']
# Данные также остались неизменными

Индексы

Документация
По дефолту Tortoise ORM использует BTree для индексации, если в поле передан параметр db_index=True или индексы поставлены в class Meta. Но если нужен другой тип индексации, как, например, FullTextIndex в MySQL, необходимо использовать tortoise.indexes.Index или его наследников.

Пример с индексами для 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"}),
        ]

Готовые индексы можно найти в модулях tortoise.contrib.mysql.indexes и tortoise.contrib.postgres.indexes

Также можно расширить tortoise.indexes.Index и прописать свой индекс.
Пример из документации:

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}"

Однако для Postgres расширение отличается, необходимо наследоваться от tortoise.contrib.postgres.indexes.PostgresSQLIndex:

class BloomIndex(PostgreSQLIndex):
    INDEX_TYPE = "BLOOM"

Функции, агрегаты и выражения

Функции, агрегаты и выражения позволяют создавать сложные SQL-запросы для получения/обработки данных в определенном виде. Tortoise ORM умеет работать со всеми из них.

Более подробно в документации функций/агрегатов и документации выражений

Tortoise ORM поддерживает следующий список функций (tortoise.functions.*):

• Trim

• Length

• Coalesce

• Lower

• Upper

• Concat

Database-specific функции для рандома:

• tortoise.contrib.mysql.functions.Rand

• tortoise.contrib.postgres.functions.Random

• tortoise.contrib.sqlite.functions.Random

Tortoise ORM поддерживает следующий список агрегатов (tortoise.functions):

• Count

• Sum

• Max

• Min

• Avg

Также Tortoise ORM позволяет расширять функции. На примере JSON_SET функции:

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 поддерживает следующий список выражений (tortoise.expressions):

• Q

• F

• SubQuery

• RawSQL (Как SubQuery только в сыром SQL)

• When

• Case

Таймзоны

Работа с таймзонами вдохновлена Django ORM, но имеет некоторые отличия. В конфиге есть два параметра use_tz и timezone, которые выставляются в Tortoise.init. Также в зависимости от БД, поведение может отличаться.

Когда use_tz=True, Tortoise ORM всегда будет хранить UTC время в базе данных вне зависимости от таймзоны.
MySQL использует поле DATETIME(6)
PostgreSQL использует TIMESTAMPZ
SQLite использует TIMESTAMP когда генерирует схемы.

Для TimeField MySQL использует TIME(6), PostgreSQL использует TIMETZ, SQLite использует TIME.

Параметр timezone определяет, какая таймзона используется при выборке из БД вне зависимости от того, какая таймзона выставлена в самой БД. Также необходимо использовать tortoise.timezone.now() вместо datetime.datetime.now().

Более подробно в документации.

Сигналы

Tortoise ORM поддерживает сигналы. На данный момент есть четыре сигнала:

• pre_save

• post_save

• pre_delete

• post_delete

Сигналы выполнены в виде декоратора. Найти их можно в модуле toroise.signals.*.
Работа с сигналами выглядит следующим образом:

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 и конфигурация подключения к БД

В Tortoise ORM можно задавать размер connection pool (min_size, max_size). В зависимости от БД, параметры конфигурации могут отличаться.

Подробнее о том, как сконфигурировать вашу/ваши БД можно в документации

В базовом примере мы используем подключение через db_url. Но можно использовать параметр config для передачи необходимых данных и более тонкой конфигурации подключений.

Усредненный конфиг будет выглядеть примерно так:

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",
            }
        },
    }
)

Логгирование

На данный момент Tortoise ORM имеет два логгера: tortoise.db_client и tortoise tortoise.db_clientлоггирует информацию об исполнении запросов, аtortoise` логгирует информацию о рантайме.

Если вы хотите контролировать поведение логов TortoiseORM, такие как дебаг SQL, вы можете сконфигугировать логи самостоятельно:

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)

Также, можно добавить цвета в логгер через pygments, подробнее в документации

Роутер для нескольких БД

Самый простой способ использовать несколько баз данных — это составить схему маршрутизации БД. Стандартная схема маршрутизации будет пытаться соединять объекты с оригинальной базой данных (то есть если объект был запрошен из базы данных foo, то сохранение объекта также будет в базу данных foo). Стандартная схема маршрутизации всегда будет использовать базу данных default, если другая не была передана.

Задать кастомный роутер очень просто: нужно просто создать класс, который будет реализовывать методы db_for_read(self, model) и 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"

Методы возвращают названия БД, которые были указаны в конфиге.

Соответственно, конфиг должен выглядеть примерно так:

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)

Страница в документации

Переопределение Object Manager

В Tortoise ORM, как и в Django ORM можно переопределять менеджер у модели. Менеджер — это интерфейс, через который создаются запросы к моделям Tortoise ORM.

Есть два способа работы с менеджерами: можно переопределить стандартный менеджер или добавить дополнительный.

Например:

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

В примере выше запросы, например, get или filter, не смогут вернуть объекты со status=0. Чтобы добиться этого, мы можем использовать стандартный менеджер в 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

Подробнее в документации.

Описание расширения полей

В Tortoise ORM можно расширять типы полей. Для этого необходимо отнаследоваться от поля определенного типа, который может быть представлен в БД.

В документации представлен пример расширения поля CharField для работы с Enum.

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

Когда наследуетесь, убедитесь что to_db_value возвращает тот же тип, что и родительский класс (в данном случае str) и что to_python_value принимает тот же тип, что и параметр value (в данном случае str)

Также как пример могу привести расширение поля для типа tstzrange из 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

Немного про Aerich (библиотека для работы с миграциями для Tortoise ORM)

У каждой уважающей себя ORM должен быть инструмент для работы с миграциями, таковым как раз и является Aerich для Tortoise ORM.

Aerich почти ничем не отличается от большинства решений в его поле деятельности. Через CLI Aerich позволяет засетапить проект под миграции, инициализировать БД, создать модели на основе таблиц в уже созданной БД, создавать миграции и их накатывать. Также, при необходимости, откатывать определенные миграции. Миграции создаются в чистом SQL.

Также, Aerich можно использовать напрямую из кода Python, если вы хотите менеджерить миграции через питоновские скрипты.

Не буду тут повторять документацию по инструменту, если интересно, то оставляю ссылку на документацию за более полной информацией.

Шаблон FastAPI + Tortoise ORM

Для того, чтобы ваше знакомство со стеком было более приятным и простым, я сделал шаблон для ваших приложений и тестов стека.

Репозиторий на Github

Репозиторий подготовлен под CookieCutter. Чтобы собрать темплейт под свой проект, установите CookieCutter и запустите команду:

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

Он не отлажен, но заводиться из коробки должен. В проекте создана простенькая структура для проекта, используется Poetry для работы с зависимостями и есть поддержка контейнеризации через докер.

Шаблон будет обновляться, буду рад вашим предложениям/замечаниям в issues.
В будущем планирую добавить работу с тестами через pytest.

Коротко о FastAPI Admin

Есть одна интересная библиотека на просторах интернета: FastAPI Admin

Создана она разработчиком long2ice, одним из контрибьюторов FastAPI. Для меня FastAPI Admin была одним из поводов пересесть с SQLAlchemy на Tortoise.
Однако, когда я пытался завести ее в своем проекте, у меня не получилось.

Не помню точно, но я остановился на том, что это занимает у меня ну слишком много времени и идею с FastAPI Admin пришлось выкинуть. Она поставилась, но выглядела и работала крайне коряво.

Возможно, спустя год после моей последней попытки, там выкатили обновления, которые позволят развернуть ее проще. Если у кого-то получилось развернуть это чудо на своем проекте, буду очень рад послушать про экспириенс в комментариях.

Заключение

Спасибо, что дочитали материал. Надеюсь, он был полезен. Очень хотелось собрать что-то внятное по Tortoise ORM, так как на русском языке источников практически нет. Делитесь в комментариях своим опытом работы с этой библиотекой. Буду рад пообщаться и послушать мнения хабраюзеров.