Шаг 05. Blueprints и CRUD-эндпоинты

🎯 Цель этапа

Создаём REST API-эндпоинты через Flask Blueprints. Каждый Blueprint — отдельный модуль со своим URL-префиксом. Все входящие данные валидируются через Pydantic (шаг 04), все ответы сериализуются через схемы Out. Закрываем callout-verify из урока 11: показываем правильную фильтрацию через request.args и ilike().

Эндпоинты этого шага

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

🔨 Шаги

1. Blueprint для вопросов

# app/api/questions.py
from flask import Blueprint, jsonify, request
from pydantic import ValidationError
from sqlalchemy import select

from app import db
from app.models import Question, Response
from app.schemas import (
    QuestionCreate,
    QuestionOut,
    ResponseCreate,
    ResponseOut,
    ErrorOut,
    MessageOut,
)

questions_bp = Blueprint("questions", __name__, url_prefix="/api/questions")


@questions_bp.route("/", methods=["GET"])
def list_questions():
    """
    GET /api/questions/ — список всех вопросов.

    Query params:
        search (str, опц.): фильтр по тексту вопроса (регистронезависимый)
        limit (int, опц.): максимальное количество результатов (по умолчанию 50)
        offset (int, опц.): смещение для пагинации (по умолчанию 0)

    Returns:
        200: {"items": [...], "total": N}
    """
    # Читаем query-параметры безопасно через request.args.get()
    search: str = request.args.get("search", "").strip()
    try:
        limit: int = int(request.args.get("limit", 50))
        offset: int = int(request.args.get("offset", 0))
    except ValueError:
        return jsonify(ErrorOut(error="Bad request", detail="limit and offset must be integers").model_dump()), 400

    stmt = select(Question).order_by(Question.created_at.desc())

    # ilike() — регистронезависимый LIKE, закрывает callout-verify из урока 11
    if search:
        stmt = stmt.where(Question.text.ilike(f"%{search}%"))

    total: int = db.session.execute(
        select(db.func.count()).select_from(stmt.subquery())
    ).scalar_one()

    questions = db.session.execute(
        stmt.limit(limit).offset(offset)
    ).scalars().all()

    return jsonify({
        "items": [QuestionOut.model_validate(q).model_dump() for q in questions],
        "total": total,
    }), 200


@questions_bp.route("/", methods=["POST"])
def create_question():
    """
    POST /api/questions/ — создать новый вопрос.

    Body: {"text": "..."}
    Returns:
        201: QuestionOut
        422: ValidationError
    """
    data = request.get_json(silent=True) or {}
    try:
        schema = QuestionCreate.model_validate(data)
    except ValidationError as exc:
        return jsonify(ErrorOut(error="Validation error", detail=str(exc)).model_dump()), 422

    question = Question(text=schema.text)
    db.session.add(question)
    db.session.commit()

    return jsonify(QuestionOut.model_validate(question).model_dump()), 201


@questions_bp.route("/<int:question_id>", methods=["GET"])
def get_question(question_id: int):
    """GET /api/questions/<id> — вопрос по ID."""
    question = db.session.get(Question, question_id)
    if not question:
        return jsonify(ErrorOut(error="Not found", detail=f"Question {question_id} not found").model_dump()), 404

    return jsonify(QuestionOut.model_validate(question).model_dump()), 200


@questions_bp.route("/<int:question_id>", methods=["PUT"])
def update_question(question_id: int):
    """PUT /api/questions/<id> — обновить текст вопроса."""
    question = db.session.get(Question, question_id)
    if not question:
        return jsonify(ErrorOut(error="Not found", detail=f"Question {question_id} not found").model_dump()), 404

    data = request.get_json(silent=True) or {}
    try:
        schema = QuestionCreate.model_validate(data)
    except ValidationError as exc:
        return jsonify(ErrorOut(error="Validation error", detail=str(exc)).model_dump()), 422

    question.text = schema.text
    db.session.commit()

    return jsonify(QuestionOut.model_validate(question).model_dump()), 200


@questions_bp.route("/<int:question_id>", methods=["DELETE"])
def delete_question(question_id: int):
    """DELETE /api/questions/<id> — удалить вопрос и все его ответы (CASCADE)."""
    question = db.session.get(Question, question_id)
    if not question:
        return jsonify(ErrorOut(error="Not found", detail=f"Question {question_id} not found").model_dump()), 404

    db.session.delete(question)
    db.session.commit()

    return jsonify(MessageOut(message=f"Question {question_id} deleted").model_dump()), 200


@questions_bp.route("/<int:question_id>/responses/", methods=["POST"])
def add_response(question_id: int):
    """
    POST /api/questions/<id>/responses/ — добавить голос.

    Body: {"is_agree": true}
    Returns:
        201: ResponseOut
        404: если вопрос не найден
        422: если данные невалидны
    """
    question = db.session.get(Question, question_id)
    if not question:
        return jsonify(ErrorOut(error="Not found", detail=f"Question {question_id} not found").model_dump()), 404

    data = request.get_json(silent=True) or {}
    try:
        schema = ResponseCreate.model_validate(data)
    except ValidationError as exc:
        return jsonify(ErrorOut(error="Validation error", detail=str(exc)).model_dump()), 422

    response = Response(question_id=question_id, is_agree=schema.is_agree)
    db.session.add(response)
    db.session.commit()

    return jsonify(ResponseOut.model_validate(response).model_dump()), 201


@questions_bp.route("/<int:question_id>/responses/", methods=["GET"])
def list_responses(question_id: int):
    """GET /api/questions/<id>/responses/ — все голоса по вопросу."""
    question = db.session.get(Question, question_id)
    if not question:
        return jsonify(ErrorOut(error="Not found", detail=f"Question {question_id} not found").model_dump()), 404

    responses = db.session.execute(
        select(Response)
        .where(Response.question_id == question_id)
        .order_by(Response.created_at.desc())
    ).scalars().all()

    return jsonify([ResponseOut.model_validate(r).model_dump() for r in responses]), 200

2. Blueprint для ответов (автономный)

# app/api/responses.py
# Опциональный автономный blueprint для /api/responses/
# (дополнительный способ просмотра всех ответов)
from flask import Blueprint, jsonify
from sqlalchemy import select

from app import db
from app.models import Response
from app.schemas import ResponseOut

responses_bp = Blueprint("responses", __name__, url_prefix="/api/responses")


@responses_bp.route("/", methods=["GET"])
def list_all_responses():
    """
    GET /api/responses/ — все ответы (для аналитики/дебага).
    Основной способ работы с ответами — через /api/questions/<id>/responses/.
    """
    responses = db.session.execute(
        select(Response).order_by(Response.created_at.desc()).limit(100)
    ).scalars().all()

    return jsonify([ResponseOut.model_validate(r).model_dump() for r in responses]), 200

3. Функция регистрации blueprints

# app/api/__init__.py
from flask import Flask


def register_blueprints(app: Flask) -> None:
    """
    Регистрирует все API-blueprints в приложении.

    Вызывается из create_app(). Централизованная функция —
    при добавлении нового blueprint достаточно добавить его здесь.
    """
    from app.api.questions import questions_bp
    from app.api.responses import responses_bp

    app.register_blueprint(questions_bp)
    app.register_blueprint(responses_bp)

4. Обновляем create_app() — подключаем blueprints

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

    with app.app_context():
        from app.models import Question, Response  # noqa: F401 — для Alembic

    # Регистрируем API blueprints
    from app.api import register_blueprints
    register_blueprints(app)

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

    return app

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

Структура URL: /api/questions/<id>/responses/

Вложенный URL (/questions/<id>/responses/) выражает отношение: ответы принадлежат конкретному вопросу. Это RESTful соглашение — ресурс «ответы» существует в контексте «вопроса». Альтернатива: POST /responses/ + {"question_id": 1} — менее явно.

Почему silent=True в get_json

Без silent=True Flask вернёт 400 Bad Request, если клиент отправил запрос без Content-Type: application/json или с невалидным JSON. С silent=True мы получаем None и можем вернуть более понятную ошибку через Pydantic.

✅ Проверка

1. Запускаем сервер

python run.py

2. Тестируем эндпоинты через curl

# Создать вопрос
curl -X POST http://localhost:5000/api/questions/ \
  -H "Content-Type: application/json" \
  -d "{\"text\": \"Согласны ли вы с удалённой работой?\"}"

# Получить все вопросы
curl http://localhost:5000/api/questions/

# Поиск по тексту (ilike)
curl "http://localhost:5000/api/questions/?search=удалённой"

# Получить вопрос по ID
curl http://localhost:5000/api/questions/1

# Добавить голос
curl -X POST http://localhost:5000/api/questions/1/responses/ \
  -H "Content-Type: application/json" \
  -d "{\"is_agree\": true}"

# Получить голоса по вопросу
curl http://localhost:5000/api/questions/1/responses/

# Удалить вопрос
curl -X DELETE http://localhost:5000/api/questions/1

3. Ожидаемые ответы

{"id": 1, "text": "Согласны ли вы с удалённой работой?", "created_at": "2025-..."}

{"items": [{"id": 1, "text": "...", "created_at": "..."}], "total": 1}

{"id": 1, "question_id": 1, "is_agree": true, "created_at": "2025-..."}

Диагностика

• 404 на /api/questions/ — blueprint не зарегистрирован. Проверьте register_blueprints() в create_app()
• 422 с длинным текстом ошибки — Pydantic отклонил входные данные. Проверьте формат JSON и поля
• 500 Internal Server Error — смотрите трейсбек в терминале с сервером

➡️ Что дальше

На следующем шаге добавим бизнес-фичу: агрегированную статистику голосов через func.count() и group_by(), а также централизованные error handlers для 404, 422 и 500 — чтобы убрать дублирование проверок из каждого эндпоинта.

• Готово: полный CRUD для вопросов, голосование, фильтрация ilike — всё работает
• Далее (шаг 06): статистика (GROUP BY + COUNT), централизованные error handlers