Урок 20 из 20 Средний 45 мин 150 XP

Проект: менеджер задач на ORM с тестами

Финал курса: собираем менеджер задач — модели со связью, CRUD, доска и тест-сьют, который проверяет всё это шестью зелёными тестами.

Редакция ПитоникиОбновлено 12 октября 2026

Двадцать уроков мы собирали по детали: движок, модели, сессии, CRUD, связи, join, агрегации, сырой SQL, тесты, миграции, продакшен, N+1. Пора сложить их в машину, которая едет. Финальный проект — менеджер задач: пара связанных моделей, слой приложения с CRUD и доской, тест-сьют, охраняющий каждое правило. Ничего нового — только всё сразу.

Начнём с программы, которую будем защищать: полный жизненный цикл записи одним скриптом. Создание, чтение, изменение, удаление — те самые четыре буквы CRUD, на которых стоит любое приложение с базой.

CRUD целиком: create, read, update, delete

from sqlalchemy import create_engine, ForeignKey, String, select, func
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column, Session, relationship

class Base(DeclarativeBase):
    pass

class User(Base):
    __tablename__ = "users"
    id: Mapped[int] = mapped_column(primary_key=True)
    name: Mapped[str] = mapped_column(String(40))
    tasks = relationship("Task", back_populates="user", order_by="Task.id")

    def __repr__(self):
        return f"User({self.name})"

class Task(Base):
    __tablename__ = "tasks"
    id: Mapped[int] = mapped_column(primary_key=True)
    title: Mapped[str] = mapped_column(String(80))
    done: Mapped[int] = mapped_column(default=0)
    user_id: Mapped[int | None] = mapped_column(ForeignKey("users.id"))
    user = relationship("User", back_populates="tasks")

    def __repr__(self):
        return f"Task({self.title!r}, done={self.done})"

engine = create_engine("sqlite:///:memory:")
Base.metadata.create_all(engine)

with Session(engine) as s:
    # C - create: автор сразу с двумя задачами
    anna = User(name="Анна", tasks=[Task(title="купить хлеб"), Task(title="полить цветы")])
    s.add(anna)
    s.commit()

    # U - update: хлеб куплен
    bread = s.scalars(select(Task).where(Task.title == "купить хлеб")).one()
    bread.done = 1
    s.commit()

    # R - read: доска задач
    for u in s.scalars(select(User).order_by(User.id)).all():
        print(u.name, "->", [(t.title, t.done) for t in u.tasks])

    # D - delete: поливать больше не нужно
    victim = s.scalars(select(Task).where(Task.title == "полить цветы")).one()
    s.delete(victim)
    s.commit()

    print("осталось задач:", s.scalar(select(func.count()).select_from(Task)))

Вывод

Анна -> [('купить хлеб', 1), ('полить цветы', 0)]
осталось задач: 1

Смотри, как много уже не нужно объяснять: конструктор принял tasks= и связь собралась сам, атрибут с commit превратился в UPDATE, delete с commit — в DELETE, а счётчик взяли агрегатом из урока про агрегации, не загружая строк. Это и есть контрольная работа всего курса — и она уже проходит.

Склад проекта

В настоящем репозитории у такого проекта четыре файла, и у каждого одна роль:

структура проекта

taskman/
  models.py       # User и Task со связью: единственный источник схемы
  conftest.py     # фикстура: свежая in-memory база на каждый тест
  test_app.py     # тест-сьют: CRUD, фильтры, связи
  app.py          # слой приложения: функции-будущие-роуты с сессиями

Схема каталога для чтения. В песочнице нет папок проекта — блок ниже создаёт те же файлы вызовами Path.write_text и запускает pytest.main; на своём компьютере эти файлы просто лежат в каталоге, а тесты запускает команда pytest.

Файл	Роль	Урок курса
models.py	модели, ForeignKey, relationship	уроки 3, 9, 10
conftest.py	фикстура session со свежей базой	урок 16
test_app.py	тесты CRUD, фильтров и связей	урок 16
app.py	функции со своими сессиями, как db.session	урок 15

Ключевое решение проекта — models.py один на всех: программу и тесты кормит одна и та же схема. Изменил модель — изменил один файл, а тесты сами напомнят, где поведение разошлось с ожиданиями. Слой приложения при этом остаётся тонким набором функций — будущими роутами, если проект переедет во Flask:

app.py: слой приложения в наброске

# app.py - каждая функция открывает свою сессию, как db.session во Flask
def add_task(title, user_id):
    with Session(engine) as s:
        task = Task(title=title, user_id=user_id)
        s.add(task)
        s.commit()
        return task.id

def finish_task(task_id):
    with Session(engine) as s:
        s.get(Task, task_id).done = 1
        s.commit()

def board():
    with Session(engine) as s:
        ...  # собрать и вернуть обычные данные - не живые объекты

Набросок для чтения: паттерн один на весь слой — открыл сессию, сделал дело, закоммитил, вернул данные. Такие функции уже встречались в уроке 15; в проекте их охраняет тест-сьют ниже.

весь проект: три файла и прогон тестов

from pathlib import Path

import pytest

Path("models.py").write_text('''\
from sqlalchemy import ForeignKey, String
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column, relationship


class Base(DeclarativeBase):
    pass


class User(Base):
    __tablename__ = "users"
    id: Mapped[int] = mapped_column(primary_key=True)
    name: Mapped[str] = mapped_column(String(40))
    tasks = relationship("Task", back_populates="user", order_by="Task.id")

    def __repr__(self):
        return f"User({self.name})"


class Task(Base):
    __tablename__ = "tasks"
    id: Mapped[int] = mapped_column(primary_key=True)
    title: Mapped[str] = mapped_column(String(80))
    done: Mapped[int] = mapped_column(default=0)
    user_id: Mapped[int | None] = mapped_column(ForeignKey("users.id"))
    user = relationship("User", back_populates="tasks")

    def __repr__(self):
        return f"Task({self.title!r}, done={self.done})"
''', encoding="utf-8")

Path("conftest.py").write_text('''\
import pytest
from sqlalchemy import create_engine
from sqlalchemy.orm import Session

from models import Base


@pytest.fixture
def session():
    engine = create_engine("sqlite://")   # свежая база в памяти на каждый тест
    Base.metadata.create_all(engine)
    with Session(engine) as s:
        yield s
    engine.dispose()
''', encoding="utf-8")

Path("test_app.py").write_text('''\
from models import Task, User


def test_create_task(session):
    session.add(Task(title="купить хлеб"))
    session.commit()
    saved = session.get(Task, 1)
    assert saved.title == "купить хлеб"
    assert saved.done == 0


def test_read_filter(session):
    session.add_all([Task(title="спеть", done=1), Task(title="станцевать")])
    session.commit()
    undone = session.query(Task).filter_by(done=0).all()
    assert [t.title for t in undone] == ["станцевать"]


def test_update_task(session):
    session.add(Task(title="полить цветы"))
    session.commit()
    session.get(Task, 1).done = 1
    session.commit()
    assert session.get(Task, 1).done == 1


def test_delete_task(session):
    session.add(Task(title="временная"))
    session.commit()
    session.delete(session.get(Task, 1))
    session.commit()
    assert session.get(Task, 1) is None


def test_task_knows_author(session):
    session.add(User(name="Анна", tasks=[Task(title="отчёт")]))
    session.commit()
    assert session.get(Task, 1).user.name == "Анна"


def test_user_knows_tasks(session):
    session.add(User(name="Борис", tasks=[Task(title="план"), Task(title="аудит")]))
    session.commit()
    assert [t.title for t in session.get(User, 1).tasks] == ["план", "аудит"]
''', encoding="utf-8")

rc = pytest.main(["test_app.py", "-q", "--no-header", "-p", "no:cacheprovider"])
print("exit:", int(rc))

Вывод

......                                                                   [100%]
6 passed in 0.09s
exit: 0

Шесть точек — шесть тестов, каждый со своей свежей базой из фикстуры conftest. Читается сьют как оглавление курса: создать и прочитать, отфильтровать, обновить, удалить, и два теста на связь в обе стороны из уроков девятого и десятого. Каждый тест — три строки действия и assert, правило AAA из курса pytest: arrange, act, assert.

в терминале проекта это та же команда

pytest -q
# ......                                                                    [100%]
# 6 passed in 0.42s

На своём компьютере файлы уже лежат на диске, и тесты запускает команда pytest из каталога проекта — без write_text и pytest.main. Песочница повторяет ровно этот прогон программно.

Доска: связи и отчёты в деле

Тесты охраняют правила, а живёт проект функциями слоя приложения — будущими роутами из урока про Flask. Доска задач, список невыполненного и сводка по статусам собираются из уже знакомых кирпичей: relationship, selectinload из прошлого урока и group_by.

доска и невыполненные

from sqlalchemy import create_engine, ForeignKey, String, select
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column, Session, relationship, selectinload

class Base(DeclarativeBase):
    pass

class User(Base):
    __tablename__ = "users"
    id: Mapped[int] = mapped_column(primary_key=True)
    name: Mapped[str] = mapped_column(String(40))
    tasks = relationship("Task", back_populates="user", order_by="Task.id")

    def __repr__(self):
        return f"User({self.name})"

class Task(Base):
    __tablename__ = "tasks"
    id: Mapped[int] = mapped_column(primary_key=True)
    title: Mapped[str] = mapped_column(String(80))
    done: Mapped[int] = mapped_column(default=0)
    user_id: Mapped[int | None] = mapped_column(ForeignKey("users.id"))
    user = relationship("User", back_populates="tasks")

    def __repr__(self):
        return f"Task({self.title!r}, done={self.done})"

engine = create_engine("sqlite:///:memory:")
Base.metadata.create_all(engine)

with Session(engine) as s:
    anna = User(name="Анна", tasks=[Task(title="купить хлеб", done=1), Task(title="написать отчёт")])
    boris = User(name="Борис", tasks=[Task(title="позвонить подрядчику"), Task(title="закрыть счёт")])
    s.add_all([anna, boris])
    s.commit()

def show_board():
    with Session(engine) as s:
        for u in s.scalars(select(User).order_by(User.id)).all():
            marks = [t.title + (" [x]" if t.done else " [ ]") for t in u.tasks]
            print(u.name, ":", "; ".join(marks))

def show_undone():
    with Session(engine) as s:
        users = s.scalars(
            select(User).order_by(User.id).options(selectinload(User.tasks))
        ).all()
        for u in users:
            undone = [t.title for t in u.tasks if not t.done]
            if undone:
                print(u.name, "-", ", ".join(undone))

show_board()
print("--- осталось сделать ---")
show_undone()

Вывод

Анна : купить хлеб [x]; написать отчёт [ ]
Борис : позвонить подрядчику [ ]; закрыть счёт [ ]
--- осталось сделать ---
Анна - написать отчёт
Борис - позвонить подрядчику, закрыть счёт

Доска построена правильно ещё и потому, что show_undone подстелил selectinload из прошлого урока: цикл по авторам с чтением задач — это ровно тот рисунок, в котором живёт N+1. Раз это правило проекта, оно заслуживает не комментарий, а тест: счётчик запросов и assert.

правило проекта: у доски два запроса, не больше

from sqlalchemy import create_engine, ForeignKey, String, select, event
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column, Session, relationship, selectinload

class Base(DeclarativeBase):
    pass

class User(Base):
    __tablename__ = "users"
    id: Mapped[int] = mapped_column(primary_key=True)
    name: Mapped[str] = mapped_column(String(40))
    tasks = relationship("Task", back_populates="user", order_by="Task.id")

    def __repr__(self):
        return f"User({self.name})"

class Task(Base):
    __tablename__ = "tasks"
    id: Mapped[int] = mapped_column(primary_key=True)
    title: Mapped[str] = mapped_column(String(80))
    user_id: Mapped[int | None] = mapped_column(ForeignKey("users.id"))
    user = relationship("User", back_populates="tasks")

engine = create_engine("sqlite:///:memory:")
Base.metadata.create_all(engine)

with Session(engine) as s:
    s.add(User(name="Анна", tasks=[Task(title="отчёт")]))
    s.add(User(name="Борис", tasks=[Task(title="план")]))
    s.commit()

selects = {"n": 0}

@event.listens_for(engine, "before_cursor_execute")
def count_selects(conn, cursor, statement, params, context, executemany):
    if statement.lstrip().upper().startswith("SELECT"):
        selects["n"] += 1

with Session(engine) as s:
    users = s.scalars(
        select(User).order_by(User.id).options(selectinload(User.tasks))
    ).all()
    board = {u.name: len(u.tasks) for u in users}

assert selects["n"] == 2, f"N+1 вернулся: {selects['n']} запросов"
print("доска:", board, "- запросов:", selects["n"])

Вывод

доска: {'Анна': 1, 'Борис': 1} - запросов: 2

Фильтр по автору и статусу — где с двумя условиями из урока про where, сводка по статусам — где с group_by, а правила целостности стережёт сама база: уникальность имён и внешний ключ поднимут IntegrityError при попытке нарушить их. Проверим все три сразу.

фильтр, сводка и целостность

from sqlalchemy import create_engine, ForeignKey, String, select, func, UniqueConstraint
from sqlalchemy.exc import IntegrityError
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column, Session, relationship

class Base(DeclarativeBase):
    pass

class User(Base):
    __tablename__ = "users"
    __table_args__ = (UniqueConstraint("name"),)
    id: Mapped[int] = mapped_column(primary_key=True)
    name: Mapped[str] = mapped_column(String(40))
    tasks = relationship("Task", back_populates="user", order_by="Task.id")

    def __repr__(self):
        return f"User({self.name})"

class Task(Base):
    __tablename__ = "tasks"
    id: Mapped[int] = mapped_column(primary_key=True)
    title: Mapped[str] = mapped_column(String(80))
    done: Mapped[int] = mapped_column(default=0)
    user_id: Mapped[int | None] = mapped_column(ForeignKey("users.id"))
    user = relationship("User", back_populates="tasks")

    def __repr__(self):
        return f"Task({self.title!r}, done={self.done})"

engine = create_engine("sqlite:///:memory:")
Base.metadata.create_all(engine)

with Session(engine) as s:
    s.add(User(name="Анна", tasks=[Task(title="купить хлеб", done=1), Task(title="написать отчёт")]))
    s.add(User(name="Борис", tasks=[Task(title="позвонить подрядчику"), Task(title="закрыть счёт")]))
    s.commit()

with Session(engine) as s:
    anna = s.scalars(select(User).where(User.name == "Анна")).one()
    rows = s.scalars(
        select(Task).where(Task.user_id == anna.id, Task.done == 0).order_by(Task.id)
    ).all()
    print("невыполненные Анны:", [t.title for t in rows])

with Session(engine) as s:
    stats = s.execute(select(Task.done, func.count()).group_by(Task.done)).all()
    print("по статусам:", sorted(stats))

with Session(engine) as s:
    s.add(User(name="Анна"))
    try:
        s.commit()
    except IntegrityError:
        print("дубликат имён не прошёл - целостность стережёт база")

Вывод

невыполненные Анны: ['написать отчёт']
по статусам: [(0, 3), (1, 1)]
дубликат имён не прошёл - целостность стережёт база

Слой приложения обязан коммитить сам

Каждая функция слоя приложения открывает свою сессию и заканчивает её commit — так работает и db.session во Flask. Если habit сессии делегировать вызывающему, половина вызовов закончится молча потерянными изменениями, а тесты это поймают только те, что проверяют чтение после записи. Правило проекта: функция, изменившая данные, сама отвечает за commit — или честно возвращает несохранённый объект, если так задумано.

Что ты теперь умеешь

Тест-сьют — это способ законсервировать всё, что ты понял: каждое правило курса превращается в assert, который не даст себе забыться. А правил этих за раздел накопилось на полный инструментовный ящик:

движок и строки подключения: от sqlite:///:memory: до postgresql+psycopg на продакшене;
декларативные модели: Mapped, mapped_column, primary_key, UniqueConstraint;
сессия и её законы: add, commit, rollback, карта идентичности, flush перед id;
запросы: select, where с операторами, order_by с limit, join, агрегации с group_by;
связи: ForeignKey на стороне «многих», relationship и back_populates в обе стороны;
массовые операции: update() и delete() с where и rowcount;
сырой SQL через text() с параметрами — без инъекций и без отказа от сессии;
тесты: фикстура со свежей in-memory базой и pytest.raises на ограничения;
миграции: зачем Alembic и почему create_all не меняет живые таблицы;
производительность: счётчик запросов, N+1 и selectinload.

Куда дальше — прямые дороги: во Flask этот же ORM встраивается в веб-приложение с роутами и формами, в FastAPI — в REST API с автодокументацией, а рядом остаётся курс про SQL и SQLite для тех, кто хочет заглянуть под капот ORM и увидеть язык, на котором база говорит на самом деле.

Проект вышел из одного скрипта, но структура у него взрослая: схема в одном месте, работа с данными в функциях, правила — под охраной тестов.

Что выведет код?

Сначала предскажи ответ в голове — это главный навык программиста.

from sqlalchemy import create_engine, String, select, func
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column, Session

class Base(DeclarativeBase):
    pass

class Task(Base):
    __tablename__ = "tasks"
    id: Mapped[int] = mapped_column(primary_key=True)
    title: Mapped[str] = mapped_column(String(80))
    done: Mapped[int] = mapped_column(default=0)

engine = create_engine("sqlite:///:memory:")
Base.metadata.create_all(engine)

with Session(engine) as s:
    s.add_all([Task(title="а"), Task(title="б", done=1)])
    s.commit()

    undone = s.scalars(select(Task).where(Task.done == 0).order_by(Task.id)).all()
    print([t.title for t in undone])

    s.delete(s.get(Task, 1))
    s.commit()
    print(s.scalar(select(func.count()).select_from(Task)))

from sqlalchemy import create_engine, ForeignKey, String
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column, Session, relationship

class Base(DeclarativeBase):
    pass

class User(Base):
    __tablename__ = "users"
    id: Mapped[int] = mapped_column(primary_key=True)
    name: Mapped[str] = mapped_column(String(40))
    tasks = relationship("Task", back_populates="user", order_by="Task.id")

class Task(Base):
    __tablename__ = "tasks"
    id: Mapped[int] = mapped_column(primary_key=True)
    title: Mapped[str] = mapped_column(String(80))
    user_id: Mapped[int | None] = mapped_column(ForeignKey("users.id"))
    user = relationship("User", back_populates="tasks")

engine = create_engine("sqlite:///:memory:")
Base.metadata.create_all(engine)

with Session(engine) as s:
    anna = User(name="Анна", tasks=[Task(title="отчёт"), Task(title="план")])
    s.add(anna)
    s.commit()
    print(len(anna.tasks), anna.tasks[0].user_id)

Проверь себя

0 / 5

1. Какую роль в проекте играет models.py?

2. Зачем проекту conftest.py?

3. Что проверяет test_task_knows_author?

4. Почему show_undone использует selectinload?

5. Где в проекте место create_all?

Карточки терминов

Запомнено: 0 / 6

Практика

Довери проекту ещё один тест: собери мини-сьют из двух файлов — board.py с моделью Task (title, done) и conftest.py с фикстурой. Первый тест на создание уже написан. Добавь test_undone_filter: две задачи — выполненная «спеть» и невыполненная «станцевать» — и проверь, что filter_by(done=0) возвращает только «станцевать». Сьют должен показать 2 passed.

practice.py

Вопросы и ответы по уроку

Как организовать проект на SQLAlchemy?

Минимальный взрослый набор: models.py со всеми моделями и связями, слой приложения из функций со своими сессиями (или db.session во Flask), conftest.py с фикстурой in-memory базы и тест-сьют. Схема живёт в одном месте, тесты используют те же модели, что и продакшен-код.

Какие тесты писать для ORM-слоя?

По одному правилу на тест: сохранение и чтение по ключу, фильтры возвращают ожидаемый список, обновление и удаление доходят до базы, связь видна в обе стороны, ограничения поднимают IntegrityError через pytest.raises. Каждый тест начинается со свежей in-memory базы из фикстуры.

Что делать дальше после курса SQLAlchemy?

Встраивать ORM в веб: во Flask это расширение Flask-SQLAlchemy с db.session на каждый запрос, в FastAPI — зависимости с сессией и REST-эндпоинты. Полезен и курс чистого SQL: понимать, какой SELECT генерирует твой запрос, — признак взрослого разработчика.

Понравился урок? Сошлитесь на него

«Тест-сьют — это способ законсервировать всё, что ты понял: каждое правило курса превращается в assert, который не даст себе забыться.»

<blockquote>«Тест-сьют — это способ законсервировать всё, что ты понял: каждое правило курса превращается в assert, который не даст себе забыться.»</blockquote>
<a href="https://pythonika.ru/sqlalchemy/urok-20/">Проект: менеджер задач на ORM с тестами — Питоника</a>

Telegram VK

SQLAlchemy · Урок 16

Тестирование ORM: pytest и in-memory база

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

pytest · Урок 20

Финальный проект: полный тест-сьют модуля работы с заказами

Выпускной: модуль заказов со скидками, классом Order, JSON и валютой — и полный сьют из шестнадцати тестов, который держит его весь целиком и гоняется одной командой.

Flask · Урок 20

Проект: сервис заметок с тестами — фабрика, blueprint и pytest

Финал расширения: собираем всё выученное в одно приложение — JSON API заметок на фабрике create_app с блюпринтом, дымовой прогон через test_client и полный pytest-набор на 200, 201, 400 и 404.

Склад проекта

Доска: связи и отчёты в деле

Что ты теперь умеешь

Понравился урок? Сошлитесь на него

Что читать дальше

Похожие уроки по темам