Peewee – лёгкая ORM

Введение

Работа с базой данных — неотъемлемая часть любой современной программы. Для Python существует множество ORM-библиотек, позволяющих удобно взаимодействовать с SQL-базами. Одной из самых простых и компактных является Peewee. Она идеально подходит для небольших проектов, скриптов и даже микросервисов.

Peewee – лёгкая ORM предлагает всё необходимое для описания моделей, выполнения запросов, работы с транзакциями и интеграции с веб-фреймворками — при этом её легко освоить и внедрить.

Принципы работы ORM и позиция Peewee

Что такое ORM

ORM (Object-Relational Mapping) — это технология, позволяющая работать с базами данных через объекты языка программирования. Вместо SQL-запросов вы создаёте и взаимодействуете с Python-классами, а ORM преобразует это в SQL.

ORM предоставляет абстракцию над SQL, что делает код более читаемым и менее подверженным ошибкам. Основные преимущества использования ORM:

• Автоматическое создание SQL-запросов
• Безопасность от SQL-инъекций
• Переносимость между разными СУБД
• Удобство работы с объектами Python

Когда выбирать Peewee

Peewee — отличный выбор в следующих случаях:

• Если нужен простой и лёгкий инструмент
• Если проект небольшой или экспериментальный
• Если важна скорость разработки без сложной конфигурации
• Для прототипирования и небольших веб-приложений
• Для настольных приложений с локальной базой данных

Установка и базовая настройка

Установка Peewee

Установить Peewee можно через pip:

pip install peewee

Для работы с различными СУБД потребуются дополнительные пакеты:

# Для PostgreSQL
pip install psycopg2-binary

# Для MySQL
pip install PyMySQL

# Для SQLite (входит в состав Python)
# дополнительных пакетов не требуется

Подключение к базе данных

Peewee поддерживает несколько типов баз данных:

from peewee import SqliteDatabase, PostgresqlDatabase, MySQLDatabase

# SQLite
db = SqliteDatabase('my_database.db')

# PostgreSQL
db = PostgresqlDatabase('mydb', user='username', password='password', host='localhost', port=5432)

# MySQL
db = MySQLDatabase('mydb', user='username', password='password', host='localhost', port=3306)

Создание моделей данных

Базовая модель

from peewee import Model, CharField, DateField, IntegerField, TextField

class BaseModel(Model):
    class Meta:
        database = db

class Author(BaseModel):
    name = CharField(max_length=100)
    birth_date = DateField()
    bio = TextField(null=True)
    
class Book(BaseModel):
    title = CharField(max_length=200)
    isbn = CharField(max_length=13, unique=True)
    pages = IntegerField()
    publication_year = IntegerField()
    author = ForeignKeyField(Author, backref='books')

Создание таблиц

db.connect()
db.create_tables([Author, Book])

Типы полей в Peewee

Основные типы полей

Специальные поля

from peewee import ForeignKeyField, ManyToManyField

class Category(BaseModel):
    name = CharField()

class Book(BaseModel):
    title = CharField()
    author = ForeignKeyField(Author, backref='books')
    categories = ManyToManyField(Category, backref='books')

# Для many-to-many связей создается промежуточная таблица
BookCategory = Book.categories.get_through_model()

Операции с базой данных (CRUD)

Создание записей

# Создание одной записи
author = Author.create(name='Фёдор Достоевский', birth_date='1821-11-11')

# Создание через конструктор
author = Author(name='Лев Толстой', birth_date='1828-09-09')
author.save()

# Массовое создание
Author.insert_many([
    {'name': 'Пушкин', 'birth_date': '1799-06-06'},
    {'name': 'Лермонтов', 'birth_date': '1814-10-15'}
]).execute()

Чтение записей

# Получение всех записей
authors = Author.select()
for author in authors:
    print(author.name)

# Получение одной записи
author = Author.get(Author.name == 'Фёдор Достоевский')

# Получение первой записи или None
author = Author.select().first()

# Получение записи по ID
author = Author.get_by_id(1)

Обновление записей

# Обновление одной записи
author = Author.get(Author.name == 'Фёдор Достоевский')
author.name = 'Ф. М. Достоевский'
author.save()

# Массовое обновление
Author.update(bio='Великий русский писатель').where(
    Author.birth_date < '1850-01-01'
).execute()

Удаление записей

# Удаление одной записи
author = Author.get(Author.name == 'Лев Толстой')
author.delete_instance()

# Массовое удаление
Author.delete().where(Author.birth_date < '1800-01-01').execute()

Фильтрация, сортировка и агрегации

Фильтрация данных

# Простые условия
authors = Author.select().where(Author.name == 'Пушкин')

# Условия с операторами
authors = Author.select().where(Author.birth_date > '1800-01-01')

# Условие LIKE
authors = Author.select().where(Author.name.contains('Фёдор'))

# Условие IN
authors = Author.select().where(Author.name.in_(['Пушкин', 'Лермонтов']))

# Составные условия
authors = Author.select().where(
    (Author.birth_date > '1800-01-01') & 
    (Author.name.contains('Лев'))
)

Сортировка

# По возрастанию
authors = Author.select().order_by(Author.name)

# По убыванию
authors = Author.select().order_by(Author.birth_date.desc())

# Множественная сортировка
authors = Author.select().order_by(Author.birth_date, Author.name)

Агрегации

from peewee import fn

# Подсчет записей
count = Author.select(fn.COUNT(Author.id)).scalar()

# Средний возраст
avg_age = Author.select(fn.AVG(Author.age)).scalar()

# Группировка
results = (Author
    .select(Author.birth_date.year, fn.COUNT(Author.id))
    .group_by(Author.birth_date.year))

Отношения между моделями

Один к одному (One-to-One)

class User(BaseModel):
    username = CharField(unique=True)
    email = CharField()

class UserProfile(BaseModel):
    user = ForeignKeyField(User, backref='profile', unique=True)
    bio = TextField()
    avatar = CharField(null=True)

Один ко многим (One-to-Many)

class Author(BaseModel):
    name = CharField()

class Book(BaseModel):
    title = CharField()
    author = ForeignKeyField(Author, backref='books')

Многие ко многим (Many-to-Many)

class Student(BaseModel):
    name = CharField()

class Course(BaseModel):
    name = CharField()
    students = ManyToManyField(Student, backref='courses')

# Создание промежуточной таблицы
StudentCourse = Student.courses.get_through_model()

Продвинутые запросы

Объединения (JOIN)

# Простое объединение
query = (Book
    .select(Book, Author)
    .join(Author)
    .where(Author.name == 'Пушкин'))

# Левое объединение
query = (Author
    .select(Author, Book)
    .join(Book, JOIN.LEFT_OUTER)
    .where(Book.title.is_null(False)))

Подзапросы

# Подзапрос в условии WHERE
prolific_authors = Author.select().where(
    Author.id.in_(
        Book.select(Book.author).group_by(Book.author).having(fn.COUNT(Book.id) > 5)
    )
)

# Подзапрос в SELECT
query = Author.select(
    Author.name,
    fn.COUNT(Book.id).alias('book_count')
).join(Book, JOIN.LEFT_OUTER).group_by(Author.id)

Сложные условия

# Использование Case
from peewee import Case

query = Author.select(
    Author.name,
    Case(None, [
        (Author.birth_date < '1800-01-01', 'Старый'),
        (Author.birth_date < '1850-01-01', 'Средний'),
    ], 'Новый').alias('period')
)

Транзакции в Peewee

Контекстный менеджер

try:
    with db.atomic():
        author = Author.create(name='Новый автор')
        book = Book.create(title='Новая книга', author=author)
        # Если произойдет ошибка, все изменения откатятся
except Exception as e:
    print(f'Ошибка: {e}')

Ручное управление транзакциями

db.begin()
try:
    Author.create(name='Автор 1')
    Author.create(name='Автор 2')
    db.commit()
except Exception:
    db.rollback()
    raise

Точки сохранения (Savepoints)

with db.atomic() as txn:
    Author.create(name='Автор 1')
    
    savepoint = txn.savepoint()
    try:
        Author.create(name='Автор 2')
        # Что-то пошло не так
        raise Exception('Ошибка!')
    except Exception:
        txn.rollback(savepoint)
    
    Author.create(name='Автор 3')

Миграции в Peewee

Использование peewee-migrate

pip install peewee-migrate

from peewee_migrate import Router

router = Router(db)

# Создание миграции
router.create('add_email_to_author')

# Применение миграций
router.run()

Ручные миграции

def migrate_add_email_column():
    migrator = SqliteMigrator(db)
    
    email_field = CharField(default='')
    migrate(
        migrator.add_column('author', 'email', email_field),
    )

Асинхронность и Peewee

Использование peewee-async

import asyncio
from peewee_async import Manager, PooledPostgresqlDatabase

database = PooledPostgresqlDatabase('test', max_connections=20)
objects = Manager(database)

async def handler():
    await objects.create(Author, name='Async Author')
    authors = await objects.execute(Author.select())
    return authors

# Запуск асинхронного кода
loop = asyncio.get_event_loop()
authors = loop.run_until_complete(handler())

Интеграция с asyncio

import asyncio
import aiosqlite
from peewee import Model, CharField, SqliteDatabase

class AsyncAuthor(Model):
    name = CharField()
    
    class Meta:
        database = SqliteDatabase(':memory:')

async def async_operations():
    # Создание таблицы
    AsyncAuthor.create_table()
    
    # Асинхронные операции требуют специальных адаптеров
    # или использования peewee-async
    pass

Интеграция с веб-фреймворками

Flask

from flask import Flask, g
from peewee import SqliteDatabase

app = Flask(__name__)
db = SqliteDatabase('app.db')

@app.before_request
def before_request():
    g.db = db
    g.db.connect()

@app.after_request
def after_request(response):
    g.db.close()
    return response

# Использование с Flask-WTF
from flask_wtf import FlaskForm
from wtforms import StringField, validators

class AuthorForm(FlaskForm):
    name = StringField('Name', [validators.Length(min=1, max=100)])
    
@app.route('/authors', methods=['GET', 'POST'])
def authors():
    form = AuthorForm()
    if form.validate_on_submit():
        Author.create(name=form.name.data)
    
    authors = Author.select()
    return render_template('authors.html', authors=authors, form=form)

FastAPI

from fastapi import FastAPI, Depends
from pydantic import BaseModel
from typing import List

app = FastAPI()

class AuthorCreate(BaseModel):
    name: str
    birth_date: str

class AuthorResponse(BaseModel):
    id: int
    name: str
    birth_date: str
    
    class Config:
        from_attributes = True

def get_db():
    db.connect()
    try:
        yield db
    finally:
        db.close()

@app.post("/authors/", response_model=AuthorResponse)
async def create_author(author: AuthorCreate, db=Depends(get_db)):
    author_obj = Author.create(**author.dict())
    return AuthorResponse.from_orm(author_obj)

@app.get("/authors/", response_model=List[AuthorResponse])
async def get_authors(db=Depends(get_db)):
    authors = Author.select()
    return [AuthorResponse.from_orm(author) for author in authors]

Django (как альтернатива Django ORM)

# settings.py
DATABASES = {
    'default': {
        'ENGINE': 'django.db.backends.sqlite3',
        'NAME': BASE_DIR / 'db.sqlite3',
    }
}

# models.py
from peewee import Model, CharField, SqliteDatabase
from django.conf import settings

db = SqliteDatabase(settings.DATABASES['default']['NAME'])

class Author(Model):
    name = CharField()
    
    class Meta:
        database = db

# views.py
from django.shortcuts import render
from django.http import JsonResponse

def authors_list(request):
    authors = list(Author.select().dicts())
    return JsonResponse(authors, safe=False)

Работа с различными СУБД

SQLite

from peewee import SqliteDatabase

# Обычный файл
db = SqliteDatabase('myapp.db')

# В памяти (для тестов)
db = SqliteDatabase(':memory:')

# С дополнительными параметрами
db = SqliteDatabase('myapp.db', pragmas={
    'journal_mode': 'wal',
    'cache_size': -1024 * 64,
    'foreign_keys': 1,
    'ignore_check_constraints': 0,
    'synchronous': 0
})

PostgreSQL

from peewee import PostgresqlDatabase

db = PostgresqlDatabase(
    'mydb',
    user='username',
    password='password',
    host='localhost',
    port=5432,
    autorollback=True,
    autocommit=True,
    options={
        'sslmode': 'require',
        'application_name': 'myapp'
    }
)

MySQL

from peewee import MySQLDatabase

db = MySQLDatabase(
    'mydb',
    user='username',
    password='password',
    host='localhost',
    port=3306,
    charset='utf8mb4',
    sql_mode='PIPES_AS_CONCAT',
    use_unicode=True,
    autocommit=True
)

Тестирование с Peewee

Настройка тестовой базы данных

import unittest
from peewee import SqliteDatabase

# Тестовая база в памяти
test_db = SqliteDatabase(':memory:')

class BaseTestCase(unittest.TestCase):
    def setUp(self):
        # Перенаправляем модели на тестовую базу
        test_db.bind([Author, Book])
        test_db.connect()
        test_db.create_tables([Author, Book])
    
    def tearDown(self):
        test_db.drop_tables([Author, Book])
        test_db.close()

class AuthorTestCase(BaseTestCase):
    def test_create_author(self):
        author = Author.create(name='Test Author', birth_date='1990-01-01')
        self.assertEqual(author.name, 'Test Author')
        self.assertEqual(Author.select().count(), 1)
    
    def test_author_books(self):
        author = Author.create(name='Test Author', birth_date='1990-01-01')
        book = Book.create(title='Test Book', author=author)
        
        self.assertEqual(author.books.count(), 1)
        self.assertEqual(book.author, author)

Фикстуры для тестов

class TestFixtures:
    @classmethod
    def create_test_data(cls):
        authors = [
            Author.create(name='Автор 1', birth_date='1990-01-01'),
            Author.create(name='Автор 2', birth_date='1985-05-15'),
        ]
        
        books = [
            Book.create(title='Книга 1', author=authors[0]),
            Book.create(title='Книга 2', author=authors[1]),
        ]
        
        return authors, books

Моки и тестирование запросов

from unittest.mock import patch, MagicMock

class DatabaseTestCase(BaseTestCase):
    @patch('peewee.SqliteDatabase.execute_sql')
    def test_query_execution(self, mock_execute):
        mock_execute.return_value = MagicMock()
        
        Author.select().execute()
        
        # Проверяем, что запрос был выполнен
        mock_execute.assert_called_once()

Расширенные возможности Peewee

Кастомные поля

from peewee import Field
import json

class JSONField(Field):
    field_type = 'text'
    
    def db_value(self, value):
        return json.dumps(value)
    
    def python_value(self, value):
        if value is not None:
            return json.loads(value)

class Config(BaseModel):
    name = CharField()
    settings = JSONField()

Валидация данных

from peewee import ValidationError

class Author(BaseModel):
    name = CharField()
    email = CharField()
    
    def validate(self):
        if '@' not in self.email:
            raise ValidationError('Invalid email format')
        
        if len(self.name) < 2:
            raise ValidationError('Name too short')
    
    def save(self, *args, **kwargs):
        self.validate()
        return super().save(*args, **kwargs)

Индексы и ограничения

class Book(BaseModel):
    title = CharField()
    isbn = CharField(unique=True)
    author = ForeignKeyField(Author)
    
    class Meta:
        database = db
        indexes = (
            # Создание индекса по нескольким полям
            (('title', 'author'), False),
            # Уникальный индекс
            (('isbn',), True),
        )
        # Ограничения на уровне таблицы
        constraints = [
            SQL('CHECK (length(title) > 0)')
        ]

Сигналы и хуки

from peewee import pre_save, post_save, pre_delete, post_delete
import datetime

@pre_save(sender=Author)
def author_pre_save(sender, instance, created):
    if created:
        instance.created_at = datetime.datetime.now()

@post_save(sender=Author)
def author_post_save(sender, instance, created):
    if created:
        print(f'Создан новый автор: {instance.name}')

@pre_delete(sender=Author)
def author_pre_delete(sender, instance):
    print(f'Удаляется автор: {instance.name}')

Производительность и оптимизация

Оптимизация запросов

# Избегайте N+1 проблемы
# Плохо
authors = Author.select()
for author in authors:
    print(author.books.count())  # Отдельный запрос для каждого автора

# Хорошо
authors = Author.select().annotate(book_count=fn.COUNT(Book.id)).join(
    Book, JOIN.LEFT_OUTER
).group_by(Author.id)

for author in authors:
    print(author.book_count)

Пагинация

def get_paginated_authors(page=1, per_page=10):
    offset = (page - 1) * per_page
    authors = Author.select().offset(offset).limit(per_page)
    total = Author.select().count()
    
    return {
        'authors': list(authors),
        'total': total,
        'page': page,
        'per_page': per_page,
        'pages': (total + per_page - 1) // per_page
    }

Пулы соединений

from peewee import PooledPostgresqlDatabase
from playhouse.pool import PooledMySQLDatabase

# PostgreSQL с пулом соединений
db = PooledPostgresqlDatabase(
    'mydb',
    max_connections=20,
    stale_timeout=300,
    user='username',
    password='password'
)

# MySQL с пулом соединений
db = PooledMySQLDatabase(
    'mydb',
    max_connections=10,
    stale_timeout=300,
    user='username',
    password='password'
)

Полная таблица методов и функций Peewee

Сравнение с другими ORM

Лучшие практики

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

# models.py
from peewee import Model, CharField, DateTimeField, ForeignKeyField
import datetime

class BaseModel(Model):
    created_at = DateTimeField(default=datetime.datetime.now)
    updated_at = DateTimeField(default=datetime.datetime.now)
    
    class Meta:
        database = db
    
    def save(self, *args, **kwargs):
        self.updated_at = datetime.datetime.now()
        return super().save(*args, **kwargs)

class Author(BaseModel):
    name = CharField(max_length=100)
    email = CharField(unique=True)
    
    def __str__(self):
        return self.name

class Book(BaseModel):
    title = CharField(max_length=200)
    author = ForeignKeyField(Author, backref='books')
    
    def __str__(self):
        return f"{self.title} by {self.author.name}"

Использование менеджеров

class AuthorManager:
    @staticmethod
    def get_active_authors():
        return Author.select().where(Author.is_active == True)
    
    @staticmethod
    def get_prolific_authors(min_books=5):
        return (Author
            .select()
            .join(Book)
            .group_by(Author)
            .having(fn.COUNT(Book.id) >= min_books))

# Использование
active_authors = AuthorManager.get_active_authors()
prolific_authors = AuthorManager.get_prolific_authors(min_books=10)

Обработка исключений

from peewee import DoesNotExist, IntegrityError

try:
    author = Author.get(Author.id == 999)
except DoesNotExist:
    print("Автор не найден")

try:
    Author.create(name="Дубликат", email="existing@email.com")
except IntegrityError:
    print("Нарушение уникальности")

Сообщество и документация

Peewee имеет активное сообщество и отличную документацию:

• Официальная документация: http://docs.peewee-orm.com/
• GitHub: https://github.com/coleifer/peewee
• PyPI: https://pypi.org/project/peewee/
• Примеры кода: https://github.com/coleifer/peewee/tree/master/examples

Библиотека активно поддерживается и обновляется. Регулярно выходят новые версии с улучшениями производительности и новыми функциями.

Часто задаваемые вопросы (FAQ)

Что такое Peewee?

Peewee — это лёгкая и простая ORM-библиотека для Python, поддерживающая SQLite, PostgreSQL и MySQL. Она предоставляет интуитивный API для работы с базами данных через Python-объекты.

Подходит ли Peewee для крупных проектов?

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

Есть ли у Peewee миграции?

Да, миграции доступны через сторонние инструменты, такие как peewee-migrate. Также можно создавать миграции вручную используя встроенные инструменты.

Поддерживает ли Peewee асинхронность?

Частично, через peewee-async и другие совместные решения. Для полноценной асинхронной работы рекомендуется использовать специализированные ORM типа Tortoise ORM.

Можно ли использовать Peewee с Flask и FastAPI?

Да, интеграция с обоими фреймворками возможна и хорошо документирована. Peewee легко интегрируется с любыми Python веб-фреймворками.

Как работает связь между моделями в Peewee?

Связи реализуются через ForeignKeyField для отношений один-ко-многим и ManyToManyField для отношений многие-ко-многим, аналогично другим ORM.

Можно ли использовать Peewee с другими СУБД кроме SQLite?

Да, Peewee поддерживает PostgreSQL, MySQL и SQLite. Для каждой СУБД существует соответствующий класс подключения.

Как обеспечить безопасность при работе с Peewee?

Peewee автоматически защищает от SQL-инъекций при использовании параметризованных запросов. Также следует валидировать входные данные и использовать аутентификацию.

Заключение

Peewee — это отличный выбор для разработчиков, которым нужна простая, но мощная ORM для Python. Она предоставляет все необходимые инструменты для работы с базами данных, оставаясь при этом легкой в изучении и использовании.

Основные преимущества Peewee:

• Простота освоения и использования
• Компактность и минимальные зависимости
• Хорошая документация и активное сообщество
• Поддержка основных СУБД
• Гибкость в настройке и расширении

Peewee идеально подходит для прототипирования, небольших приложений, скриптов автоматизации и проектов, где важна скорость разработки. Для крупных проектов с высокими требованиями к производительности стоит рассмотреть более мощные решения, такие как SQLAlchemy.