Шаг 03. Модели и связи

🎯 Цель этапа

Создаём ORM-модели для двух сущностей: Question (вопрос) и Response (голос — «за» или «против»). Используем современный декларативный синтаксис SQLAlchemy 2.x с аннотациями типов: Mapped[T] вместо старых Column(). Закрываем callout-verify из урока 09.

После этого шага у нас будет

• Модель Question: id, text, created_at, relationship к Response
• Модель Response: id, question_id (FK), is_agree, created_at
• Правильная связь один-ко-многим через relationship()
• Таблицы в БД, созданные через миграцию (не через create_all())

📄 Затрагиваемые файлы

🔨 Шаги

1. Создаём модель Question

# app/models/question.py
from __future__ import annotations

from datetime import datetime
from typing import TYPE_CHECKING

from sqlalchemy import String, func
from sqlalchemy.orm import Mapped, mapped_column, relationship

from app import db

if TYPE_CHECKING:
    # Импортируем только для type checker, чтобы не было цикличного импорта
    from app.models.response import Response


class Question(db.Model):
    """
    Вопрос для голосования сообщества.

    Attributes:
        id: уникальный идентификатор
        text: текст вопроса (обязательный, не пустой)
        created_at: время создания (заполняется автоматически сервером БД)
        responses: все голоса по этому вопросу (lazy='dynamic' для больших выборок)
    """
    __tablename__ = "questions"

    id: Mapped[int] = mapped_column(primary_key=True, autoincrement=True)
    text: Mapped[str] = mapped_column(String(500), nullable=False)
    created_at: Mapped[datetime] = mapped_column(
        default=func.now(),
        server_default=func.now(),
    )

    # Связь один-ко-многим: один вопрос — много ответов
    # cascade="all, delete-orphan": при удалении вопроса удаляются все его ответы
    responses: Mapped[list["Response"]] = relationship(
        "Response",
        back_populates="question",
        cascade="all, delete-orphan",
        lazy="select",
    )

    def __repr__(self) -> str:
        return f"<Question id={self.id} text={self.text[:30]!r}>"

2. Создаём модель Response

# app/models/response.py
from __future__ import annotations

from datetime import datetime
from typing import TYPE_CHECKING

from sqlalchemy import ForeignKey, func
from sqlalchemy.orm import Mapped, mapped_column, relationship

from app import db

if TYPE_CHECKING:
    from app.models.question import Question


class Response(db.Model):
    """
    Голос пользователя по вопросу: «за» (True) или «против» (False).

    Attributes:
        id: уникальный идентификатор
        question_id: FK на таблицу questions (ON DELETE CASCADE)
        is_agree: True = согласен, False = не согласен
        created_at: время голосования (авто)
        question: обратная ссылка на вопрос
    """
    __tablename__ = "responses"

    id: Mapped[int] = mapped_column(primary_key=True, autoincrement=True)
    question_id: Mapped[int] = mapped_column(
        ForeignKey("questions.id", ondelete="CASCADE"),
        nullable=False,
        index=True,    # индекс для быстрых запросов по question_id
    )
    is_agree: Mapped[bool] = mapped_column(nullable=False)
    created_at: Mapped[datetime] = mapped_column(
        default=func.now(),
        server_default=func.now(),
    )

    # Обратная ссылка на вопрос
    question: Mapped["Question"] = relationship(
        "Question",
        back_populates="responses",
    )

    def __repr__(self) -> str:
        vote = "agree" if self.is_agree else "disagree"
        return f"<Response id={self.id} q={self.question_id} vote={vote}>"

3. Обновляем app/models/__init__.py

# app/models/__init__.py
# Экспортируем модели для удобного импорта в других модулях:
#   from app.models import Question, Response
from app.models.question import Question
from app.models.response import Response

__all__ = ["Question", "Response"]

4. Импортируем модели в create_app() для Alembic

Alembic (Flask-Migrate) автоматически обнаруживает модели только если они импортированы до вызова migrate. Добавляем импорт в create_app():

# app/__init__.py
from flask import Flask, jsonify
from flask_sqlalchemy import SQLAlchemy
from flask_migrate import Migrate

from config import config_map, Config

db: SQLAlchemy = SQLAlchemy()
migrate: Migrate = Migrate()


def create_app(config_name: str = "development") -> Flask:
    app = Flask(__name__)
    config_class: type[Config] = config_map[config_name]
    app.config.from_object(config_class)

    db.init_app(app)
    migrate.init_app(app, db)

    # ВАЖНО: импортируем модели ВНУТРИ create_app(), после init_app().
    # Это позволяет Alembic видеть таблицы при генерации миграций.
    with app.app_context():
        from app.models import Question, Response  # noqa: F401

    @app.route("/health")
    def health_check():
        return jsonify({"status": "ok", "env": config_name}), 200

    return app

5. Создаём и применяем миграцию

# Генерируем миграцию — Alembic обнаружит новые таблицы
flask db migrate -m "add questions and responses tables"

# Просматриваем сгенерированный файл (убеждаемся, что таблицы есть)
# migrations/versions/xxxx_add_questions_and_responses_tables.py

# Применяем
flask db upgrade

🧠 Объяснение логики

Старый vs новый синтаксис SQLAlchemy

Почему cascade="all, delete-orphan"

Если вопрос удаляется, все его голоса теряют смысл. Без каскада они останутся в таблице с question_id, указывающим на несуществующую запись — «осиротевшие» записи. Каскад гарантирует целостность: удалили вопрос — удалились все его ответы.

back_populates vs backref

В SQLAlchemy 2.x рекомендуется back_populates вместо backref. back_populates требует явного объявления с обеих сторон, но зато каждая сторона видна в коде — type checker понимает типы, IDE подсказывает методы.

✅ Проверка

1. Проверяем миграцию

flask db migrate -m "add questions and responses tables"
flask db upgrade

2. Проверяем таблицы через flask shell

flask shell

# В интерактивной оболочке:
from app.models import Question, Response
from app import db

# Создаём тестовый вопрос
q = Question(text="Согласны ли вы с удалённой работой?")
db.session.add(q)
db.session.commit()
print(q.id)  # должно вывести 1

# Добавляем ответ
r = Response(question_id=q.id, is_agree=True)
db.session.add(r)
db.session.commit()

# Проверяем связь
print(q.responses)  # [<Response id=1 q=1 vote=agree>]
print(r.question)   # <Question id=1 text='Согласны ли вы...'>

exit()

3. Ожидаемый результат

INFO  [alembic.runtime.migration] Running upgrade xxxx -> yyyy, add questions and responses tables

Диагностика

• Миграция пустая (нет op.create_table) — модели не импортированы в create_app(). Проверьте импорт.
• ImportError: cannot import name 'Question' — проверьте app/models/__init__.py
• CircularImportError — используйте TYPE_CHECKING guard для перекрёстных ссылок

➡️ Что дальше

На следующем шаге создаём Pydantic v2 схемы для валидации входящих данных и сериализации ORM-объектов в JSON. Используем model_config = ConfigDict(from_attributes=True) для чтения атрибутов SQLAlchemy-моделей.

• Готово: модели Question и Response, связь один-ко-многим, таблицы в БД через миграцию
• Далее (шаг 04): Pydantic v2 схемы валидации и сериализации