Tortoise-ORM: полное руководство по асинхронной работе с базами данных в Python
Введение
Современная разработка на Python все чаще требует асинхронного подхода, особенно при создании высокопроизводительных веб-приложений и API. Традиционные ORM-решения часто не поддерживают асинхронность из коробки, что создает проблемы для разработчиков, работающих с современными фреймворками вроде FastAPI, Starlette или Quart.
Tortoise-ORM решает эту проблему, предоставляя мощную асинхронную ORM-библиотеку, которая нативно поддерживает async/await паттерны. Эта библиотека вдохновлена Django ORM, что делает ее освоение простым для разработчиков с опытом работы с Django, но при этом она специально создана для асинхронных приложений.
Что такое Tortoise-ORM и зачем она нужна
Основы ORM и преимущества асинхронности
ORM (Object-Relational Mapping) представляет собой технологию программирования, которая позволяет описывать структуры баз данных в виде классов и объектов в коде. Вместо написания SQL-запросов вручную, разработчики могут работать с данными через удобные Python-объекты.
Асинхронность в ORM критически важна для современных приложений, поскольку позволяет:
- Обрабатывать тысячи одновременных запросов без блокировки
- Эффективно использовать ресурсы сервера
- Создавать масштабируемые API с высокой производительностью
- Избегать блокировок при длительных операциях с базой данных
Основные преимущества Tortoise-ORM
Tortoise-ORM предоставляет ряд ключевых преимуществ:
Нативная асинхронность: Библиотека создана с нуля для работы с async/await, что обеспечивает максимальную производительность.
Знакомый синтаксис: API основан на Django ORM, что упрощает миграцию и обучение.
Широкая поддержка СУБД: Работает с PostgreSQL, MySQL, SQLite через асинхронные драйверы.
Интеграция с современными фреймворками: Простая интеграция с FastAPI, Starlette и другими ASGI-приложениями.
Сферы применения Tortoise-ORM
Веб-разработка и API
Tortoise-ORM идеально подходит для создания RESTful API и GraphQL-сервисов на базе FastAPI или Starlette. Асинхронная природа библиотеки позволяет обрабатывать множество одновременных запросов к базе данных без деградации производительности.
Микросервисная архитектура
В микросервисных архитектурах, где важна скорость обработки запросов и эффективное использование ресурсов, Tortoise-ORM обеспечивает необходимую производительность и масштабируемость.
Высоконагруженные приложения
Для приложений с высокой нагрузкой, где требуется обработка большого количества одновременных подключений к базе данных, асинхронная природа Tortoise-ORM становится критически важной.
Установка и первоначальная настройка
Базовая установка
pip install tortoise-orm
Установка с поддержкой конкретных СУБД
Для PostgreSQL:
pip install tortoise-orm[asyncpg]
Для MySQL:
pip install tortoise-orm[aiomysql]
Для SQLite:
pip install tortoise-orm[aiosqlite]
Полная установка со всеми зависимостями
pip install tortoise-orm[asyncpg,aiomysql,aiosqlite]
Создание и настройка моделей
Определение базовых моделей
Создание моделей в Tortoise-ORM происходит через наследование от базового класса Model:
from tortoise import fields
from tortoise.models import Model
class Author(Model):
id = fields.IntField(pk=True)
name = fields.CharField(max_length=100)
birth_date = fields.DateField(null=True)
biography = fields.TextField(null=True)
is_active = fields.BooleanField(default=True)
created_at = fields.DatetimeField(auto_now_add=True)
updated_at = fields.DatetimeField(auto_now=True)
class Meta:
table = "authors"
ordering = ["name"]
Расширенные типы полей
Tortoise-ORM поддерживает широкий спектр типов полей для различных задач:
class Book(Model):
id = fields.IntField(pk=True)
title = fields.CharField(max_length=200)
isbn = fields.CharField(max_length=13, unique=True)
page_count = fields.IntField()
price = fields.DecimalField(max_digits=10, decimal_places=2)
publication_date = fields.DateField()
description = fields.TextField()
cover_image = fields.CharField(max_length=500, null=True)
metadata = fields.JSONField(default=dict)
author = fields.ForeignKeyField('models.Author', related_name='books')
genres = fields.ManyToManyField('models.Genre', related_name='books')
class Meta:
table = "books"
unique_together = (("title", "author"),)
Инициализация и подключение к базе данных
Базовая инициализация
from tortoise import Tortoise
async def init_db():
await Tortoise.init(
db_url='sqlite://db.sqlite3',
modules={'models': ['models']}
)
await Tortoise.generate_schemas()
async def close_db():
await Tortoise.close_connections()
Конфигурация для различных СУБД
# PostgreSQL
DATABASE_URL = "postgres://user:password@localhost:5432/database"
# MySQL
DATABASE_URL = "mysql://user:password@localhost:3306/database"
# SQLite
DATABASE_URL = "sqlite://./database.sqlite3"
# Конфигурация через словарь
TORTOISE_CONFIG = {
"connections": {
"default": {
"engine": "tortoise.backends.asyncpg",
"credentials": {
"host": "localhost",
"port": "5432",
"user": "postgres",
"password": "password",
"database": "mydb",
}
}
},
"apps": {
"models": {
"models": ["app.models", "aerich.models"],
"default_connection": "default",
}
},
}
Основные операции с данными
Создание записей
# Создание одной записи
author = await Author.create(
name='Федор Достоевский',
birth_date=date(1821, 11, 11),
biography='Великий русский писатель'
)
# Создание нескольких записей
authors = await Author.bulk_create([
Author(name='Лев Толстой', birth_date=date(1828, 9, 9)),
Author(name='Anton Чехов', birth_date=date(1860, 1, 29)),
])
Получение данных
# Получение одной записи
author = await Author.get(name='Федор Достоевский')
# Получение с условием или исключение
try:
author = await Author.get(id=999)
except DoesNotExist:
print("Автор не найден")
# Получение или создание
author, created = await Author.get_or_create(
name='Новый автор',
defaults={'birth_date': date(1900, 1, 1)}
)
# Получение всех записей
all_authors = await Author.all()
# Получение первой/последней записи
first_author = await Author.first()
last_author = await Author.last()
Фильтрация и поиск
# Простая фильтрация
active_authors = await Author.filter(is_active=True)
# Сложная фильтрация
authors = await Author.filter(
birth_date__gte=date(1800, 1, 1),
birth_date__lt=date(1900, 1, 1)
)
# Поиск по части текста
authors = await Author.filter(name__icontains='толстой')
# Исключение записей
authors = await Author.exclude(birth_date__isnull=True)
# Комбинированные условия
from tortoise.queryset import Q
authors = await Author.filter(
Q(name__icontains='толстой') | Q(name__icontains='достоев')
)
Обновление данных
# Обновление одной записи
author = await Author.get(id=1)
author.name = 'Ф. М. Достоевский'
await author.save()
# Обновление из словаря
await author.update_from_dict({'name': 'Новое имя', 'is_active': False})
await author.save()
# Массовое обновление
await Author.filter(birth_date__lt=date(1850, 1, 1)).update(is_active=False)
Удаление данных
# Удаление одной записи
author = await Author.get(id=1)
await author.delete()
# Массовое удаление
await Author.filter(is_active=False).delete()
Работа со связями между моделями
Определение связей
class Genre(Model):
id = fields.IntField(pk=True)
name = fields.CharField(max_length=50, unique=True)
class Publisher(Model):
id = fields.IntField(pk=True)
name = fields.CharField(max_length=100)
country = fields.CharField(max_length=50)
class Book(Model):
id = fields.IntField(pk=True)
title = fields.CharField(max_length=200)
# Один-ко-многим
author = fields.ForeignKeyField('models.Author', related_name='books')
publisher = fields.ForeignKeyField('models.Publisher', related_name='books')
# Многие-ко-многим
genres = fields.ManyToManyField('models.Genre', related_name='books')
class AuthorProfile(Model):
id = fields.IntField(pk=True)
website = fields.CharField(max_length=200, null=True)
social_media = fields.JSONField(default=dict)
# Один-к-одному
author = fields.OneToOneField('models.Author', related_name='profile')
Работа с ForeignKey
# Создание с связью
author = await Author.create(name='Пушкин')
book = await Book.create(
title='Евгений Онегин',
author=author
)
# Получение связанных данных
book = await Book.get(id=1)
author = await book.author # Асинхронное получение
# Обратная связь
author = await Author.get(id=1)
books = await author.books.all()
Работа с ManyToMany
# Создание и добавление связей
book = await Book.create(title='Новая книга', author=author)
genre1 = await Genre.create(name='Роман')
genre2 = await Genre.create(name='Классика')
await book.genres.add(genre1, genre2)
# Получение связанных записей
genres = await book.genres.all()
# Удаление связей
await book.genres.remove(genre1)
await book.genres.clear() # Удалить все связи
Оптимизация запросов
Select Related и Prefetch Related
# select_related для ForeignKey (JOIN)
books = await Book.all().select_related('author', 'publisher')
# prefetch_related для ManyToMany и reverse ForeignKey
authors = await Author.all().prefetch_related('books', 'books__genres')
# Комбинированное использование
books = await Book.all().select_related('author').prefetch_related('genres')
Аггрегация и аннотации
from tortoise.functions import Count, Sum, Avg
# Подсчет количества книг у каждого автора
authors = await Author.all().annotate(
book_count=Count('books')
).filter(book_count__gt=0)
# Средняя цена книг по жанрам
genres = await Genre.all().annotate(
avg_price=Avg('books__price')
)
Транзакции и целостность данных
Использование транзакций
from tortoise.transactions import in_transaction
async def create_book_with_author():
async with in_transaction():
author = await Author.create(name='Новый автор')
book = await Book.create(title='Новая книга', author=author)
return book
# Обработка ошибок в транзакциях
async def safe_create():
try:
async with in_transaction():
# Операции с базой данных
pass
except Exception as e:
# Транзакция автоматически откатывается
print(f"Ошибка: {e}")
Атомарные операции
from tortoise.transactions import atomic
@atomic()
async def update_book_info(book_id: int, new_title: str):
book = await Book.get(id=book_id)
book.title = new_title
await book.save()
# Операция выполнится атомарно
Система миграций с Aerich
Установка и настройка Aerich
pip install aerich
Инициализация миграций
aerich init -t settings.TORTOISE_ORM
aerich init-db
Создание и применение миграций
# Создание миграции
aerich migrate --name "add_new_fields"
# Применение миграций
aerich upgrade
# Откат миграций
aerich downgrade
# Просмотр истории миграций
aerich history
Конфигурация для Aerich
# settings.py
TORTOISE_ORM = {
"connections": {"default": "sqlite://db.sqlite3"},
"apps": {
"models": {
"models": ["app.models", "aerich.models"],
"default_connection": "default",
},
},
}
Интеграция с FastAPI
Базовая настройка
from fastapi import FastAPI
from tortoise.contrib.fastapi import register_tortoise
app = FastAPI()
register_tortoise(
app,
db_url="sqlite://./test.db",
modules={"models": ["app.models"]},
generate_schemas=True,
add_exception_handlers=True,
)
Создание Pydantic схем
from tortoise.contrib.pydantic import pydantic_model_creator
# Создание схем для API
Author_Pydantic = pydantic_model_creator(Author, name="Author")
AuthorIn_Pydantic = pydantic_model_creator(
Author, name="AuthorIn", exclude_readonly=True
)
# Использование в эндпоинтах
@app.post("/authors/", response_model=Author_Pydantic)
async def create_author(author: AuthorIn_Pydantic):
author_obj = await Author.create(**author.dict())
return await Author_Pydantic.from_tortoise_orm(author_obj)
Расширенная интеграция
from contextlib import asynccontextmanager
@asynccontextmanager
async def lifespan(app: FastAPI):
# Инициализация при запуске
await Tortoise.init(
db_url="sqlite://./test.db",
modules={"models": ["app.models"]}
)
await Tortoise.generate_schemas()
yield
# Очистка при завершении
await Tortoise.close_connections()
app = FastAPI(lifespan=lifespan)
Поддерживаемые системы управления базами данных
PostgreSQL
# Настройка для PostgreSQL
TORTOISE_CONFIG = {
"connections": {
"default": {
"engine": "tortoise.backends.asyncpg",
"credentials": {
"host": "localhost",
"port": "5432",
"user": "postgres",
"password": "password",
"database": "mydb",
"minsize": 1,
"maxsize": 10,
}
}
}
}
MySQL
# Настройка для MySQL
TORTOISE_CONFIG = {
"connections": {
"default": {
"engine": "tortoise.backends.mysql",
"credentials": {
"host": "localhost",
"port": "3306",
"user": "root",
"password": "password",
"database": "mydb",
"minsize": 1,
"maxsize": 10,
}
}
}
}
SQLite
# Настройка для SQLite
TORTOISE_CONFIG = {
"connections": {
"default": "sqlite://./database.sqlite3"
}
}
Тестирование приложений с Tortoise-ORM
Настройка тестовой среды
import pytest
from tortoise.contrib.test import finalizer, initializer
@pytest.fixture(scope="session", autouse=True)
async def initialize_tests():
await initializer(
["app.models"],
db_url="sqlite://:memory:",
app_label="models",
)
yield
await finalizer()
# Тест с изоляцией
@pytest.mark.asyncio
async def test_create_author():
author = await Author.create(name="Тестовый автор")
assert author.name == "Тестовый автор"
assert author.id is not None
Использование транзакций в тестах
from tortoise.transactions import in_transaction
@pytest.mark.asyncio
async def test_with_transaction():
async with in_transaction():
author = await Author.create(name="Тест")
book = await Book.create(title="Тест книга", author=author)
assert await Book.filter(author=author).count() == 1
Производительность и оптимизация
Индексы и оптимизация запросов
class OptimizedModel(Model):
id = fields.IntField(pk=True)
name = fields.CharField(max_length=100, index=True)
email = fields.CharField(max_length=100, unique=True)
created_at = fields.DatetimeField(auto_now_add=True, index=True)
class Meta:
table = "optimized_model"
indexes = [
["name", "created_at"], # Составной индекс
]
Пакетные операции
# Пакетное создание
authors = await Author.bulk_create([
Author(name=f"Автор {i}") for i in range(100)
])
# Пакетное обновление
await Author.filter(is_active=True).update(updated_at=datetime.now())
Кэширование запросов
from tortoise.queryset import QuerySet
class CachedQuerySet(QuerySet):
def __init__(self, model, using_db=None):
super().__init__(model, using_db)
self._cache = {}
async def get(self, *args, **kwargs):
cache_key = str(kwargs)
if cache_key in self._cache:
return self._cache[cache_key]
result = await super().get(*args, **kwargs)
self._cache[cache_key] = result
return result
Справочник методов и функций Tortoise-ORM
Таблица основных методов модели
| Метод | Описание | Пример использования |
|---|---|---|
create(**kwargs) |
Создание новой записи | await Author.create(name="Имя") |
get(**kwargs) |
Получение одной записи | await Author.get(id=1) |
get_or_create(**kwargs) |
Получение или создание | await Author.get_or_create(name="Имя") |
filter(**kwargs) |
Фильтрация записей | await Author.filter(is_active=True) |
exclude(**kwargs) |
Исключение записей | await Author.exclude(is_active=False) |
all() |
Получение всех записей | await Author.all() |
first() |
Первая запись | await Author.first() |
last() |
Последняя запись | await Author.last() |
count() |
Количество записей | await Author.all().count() |
exists() |
Проверка существования | await Author.filter(name="Имя").exists() |
update(**kwargs) |
Массовое обновление | await Author.filter(id=1).update(name="Новое имя") |
delete() |
Удаление записей | await Author.filter(id=1).delete() |
bulk_create(objects) |
Пакетное создание | await Author.bulk_create([Author(name="1"), Author(name="2")]) |
bulk_update(objects, fields) |
Пакетное обновление | await Author.bulk_update(authors, ["name"]) |
select_related(*args) |
Жадная загрузка FK | await Book.all().select_related("author") |
prefetch_related(*args) |
Жадная загрузка M2M | await Author.all().prefetch_related("books") |
annotate(**kwargs) |
Аннотации | await Author.all().annotate(book_count=Count("books")) |
order_by(*args) |
Сортировка | await Author.all().order_by("name") |
limit(count) |
Ограничение количества | await Author.all().limit(10) |
offset(count) |
Смещение | await Author.all().offset(10) |
distinct() |
Уникальные записи | await Author.all().distinct() |
values(*args) |
Получение значений полей | await Author.all().values("name", "id") |
values_list(*args) |
Получение значений в виде списка | await Author.all().values_list("name", flat=True) |
only(*args) |
Получение только указанных полей | await Author.all().only("name") |
defer(*args) |
Отложенная загрузка полей | await Author.all().defer("biography") |
Таблица операторов фильтрации
| Оператор | Описание | Пример |
|---|---|---|
exact |
Точное совпадение | filter(name__exact="Имя") |
iexact |
Точное совпадение (без учета регистра) | filter(name__iexact="имя") |
contains |
Содержит подстроку | filter(name__contains="Толст") |
icontains |
Содержит подстроку (без учета регистра) | filter(name__icontains="толст") |
startswith |
Начинается с | filter(name__startswith="А") |
istartswith |
Начинается с (без учета регистра) | filter(name__istartswith="а") |
endswith |
Заканчивается на | filter(name__endswith="ский") |
iendswith |
Заканчивается на (без учета регистра) | filter(name__iendswith="ский") |
gt |
Больше | filter(age__gt=18) |
gte |
Больше или равно | filter(age__gte=18) |
lt |
Меньше | filter(age__lt=65) |
lte |
Меньше или равно | filter(age__lte=65) |
in |
В списке значений | filter(id__in=[1, 2, 3]) |
not_in |
Не в списке значений | filter(id__not_in=[1, 2, 3]) |
isnull |
Значение NULL | filter(birth_date__isnull=True) |
not_isnull |
Значение не NULL | filter(birth_date__not_isnull=True) |
range |
В диапазоне | filter(age__range=[18, 65]) |
year |
Год из даты | filter(birth_date__year=1990) |
month |
Месяц из даты | filter(birth_date__month=12) |
day |
День из даты | filter(birth_date__day=25) |
Таблица типов полей
| Тип поля | Описание | Параметры |
|---|---|---|
IntField |
Целое число | pk=False |
BigIntField |
Большое целое число | pk=False |
SmallIntField |
Малое целое число | pk=False |
CharField |
Строка | max_length, unique=False, index=False |
TextField |
Длинный текст | pk=False |
BooleanField |
Логическое значение | default=False |
DateField |
Дата | auto_now=False, auto_now_add=False |
DatetimeField |
Дата и время | auto_now=False, auto_now_add=False |
TimeField |
Время | auto_now=False, auto_now_add=False |
DecimalField |
Десятичное число | max_digits, decimal_places |
FloatField |
Число с плавающей точкой | |
JSONField |
JSON данные | default=dict |
UUIDField |
UUID | default=uuid.uuid4 |
BinaryField |
Двоичные данные | |
ForeignKeyField |
Внешний ключ | model_name, related_name, on_delete |
OneToOneField |
Связь один-к-одному | model_name, related_name, on_delete |
ManyToManyField |
Связь многие-ко-многим | model_name, related_name, through |
Сравнение с другими ORM
Сравнительная таблица популярных ORM
| Характеристика | Tortoise-ORM | SQLAlchemy + asyncpg | Django ORM | Peewee |
|---|---|---|---|---|
| Асинхронность | Нативная | Поддерживается | Нет | Частичная |
| Простота изучения | Высокая | Средняя | Высокая | Высокая |
| Производительность | Высокая | Высокая | Средняя | Средняя |
| Поддержка миграций | Через Aerich | Alembic | Встроенная | peewee-migrate |
| Размер сообщества | Растущее | Большое | Огромное | Среднее |
| Документация | Хорошая | Отличная | Отличная | Хорошая |
| Подходит для FastAPI | Идеально | Хорошо | Нет | Средне |
| Поддержка типов | Хорошая | Отличная | Отличная | Хорошая |
Лучшие практики и рекомендации
Структура проекта
project/
├── app/
│ ├── __init__.py
│ ├── models/
│ │ ├── __init__.py
│ │ ├── author.py
│ │ ├── book.py
│ │ └── base.py
│ ├── schemas/
│ │ ├── __init__.py
│ │ ├── author.py
│ │ └── book.py
│ ├── services/
│ │ ├── __init__.py
│ │ ├── author_service.py
│ │ └── book_service.py
│ └── config.py
├── migrations/
├── tests/
└── main.py
Использование сервисного слоя
# services/author_service.py
class AuthorService:
@staticmethod
async def create_author(name: str, birth_date: date = None) -> Author:
return await Author.create(name=name, birth_date=birth_date)
@staticmethod
async def get_author_with_books(author_id: int) -> Author:
return await Author.get(id=author_id).prefetch_related('books')
@staticmethod
async def get_popular_authors(min_books: int = 5) -> List[Author]:
return await Author.all().annotate(
book_count=Count('books')
).filter(book_count__gte=min_books)
Обработка ошибок
from tortoise.exceptions import DoesNotExist, IntegrityError
async def safe_get_author(author_id: int) -> Optional[Author]:
try:
return await Author.get(id=author_id)
except DoesNotExist:
return None
except IntegrityError as e:
logger.error(f"Integrity error: {e}")
raise
Часто задаваемые вопросы
Что такое Tortoise-ORM и чем она отличается от других ORM?
Tortoise-ORM — это асинхронная ORM-библиотека для Python, созданная специально для работы с async/await паттернами. Основное отличие от традиционных ORM заключается в нативной поддержке асинхронности, что делает ее идеальной для современных веб-приложений на FastAPI, Starlette и других ASGI-фреймворках.
Поддерживает ли Tortoise-ORM все основные системы управления базами данных?
Да, Tortoise-ORM поддерживает PostgreSQL, MySQL и SQLite через соответствующие асинхронные драйверы (asyncpg, aiomysql, aiosqlite). Для каждой СУБД доступны специфические возможности, но основной API остается единообразным.
Как выполнять миграции в Tortoise-ORM?
Tortoise-ORM не имеет встроенной системы миграций, но тесно интегрирована с инструментом Aerich. Aerich позволяет создавать, применять и откатывать миграции, а также отслеживать изменения в структуре базы данных.
Можно ли использовать Tortoise-ORM в продакшене?
Да, Tortoise-ORM активно используется в производственных системах. Библиотека стабильна, хорошо протестирована и обеспечивает высокую производительность. Важно правильно настроить пулы соединений и использовать оптимизации запросов.
Как интегрировать Tortoise-ORM с Pydantic в FastAPI?
Tortoise-ORM предоставляет встроенную поддержку генерации Pydantic-схем через pydantic_model_creator. Это позволяет автоматически создавать схемы валидации для API на основе моделей базы данных.
Поддерживает ли Tortoise-ORM транзакции?
Да, Tortoise-ORM полностью поддерживает транзакции через контекстные менеджеры in_transaction() и декораторы @atomic(). Это обеспечивает целостность данных при выполнении связанных операций.
Как оптимизировать производительность запросов?
Для оптимизации производительности используйте select_related() для ForeignKey-связей, prefetch_related() для ManyToMany-связей, создавайте индексы для часто используемых полей и применяйте пакетные операции для массовых изменений.
Можно ли использовать Tortoise-ORM с другими фреймворками кроме FastAPI?
Да, Tortoise-ORM совместима с любыми ASGI-фреймворками, включая Starlette, Quart, Sanic и другие. Главное требование — поддержка асинхронности на уровне фреймворка.
Полезные ресурсы и документация
Официальные ресурсы
- Официальная документация: https://tortoise-orm.readthedocs.io/
- GitHub репозиторий: https://github.com/tortoise/tortoise-orm
- PyPI страница: https://pypi.org/project/tortoise-orm/
Дополнительные инструменты
- Aerich (миграции): https://github.com/tortoise/aerich
- Tortoise-CLI (утилиты командной строки): https://github.com/tortoise/tortoise-cli
- FastAPI интеграция: https://github.com/tortoise/fastapi-tortoise
Сообщество и поддержка
- Discord сервер: Активное сообщество разработчиков
- GitHub Issues: Отслеживание багов и запросов функций
- Stack Overflow: Тег
tortoise-ormдля вопросов
Tortoise-ORM представляет собой мощное и современное решение для асинхронной работы с базами данных в Python. Благодаря простому API, высокой производительности и отличной интеграции с современными фреймворками, эта библиотека становится идеальным выбором для создания масштабируемых веб-приложений и API.
Настоящее и будущее развития ИИ: классической математики уже недостаточно
Эксперты предупредили о рисках фейковой благотворительности с помощью ИИ
В России разработали универсального ИИ-агента для роботов и индустриальных процессов