Шаг 06. Статистика и централизованная обработка ошибок

🎯 Цель этапа

Добавляем две вещи, которые делают API production-ready: бизнес-статистику через SQL-агрегации и централизованную обработку ошибок через Flask error handlers.

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

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

🔨 Шаги

1. Добавляем схему статистики

# app/schemas/common.py
from pydantic import BaseModel, ConfigDict


class ErrorOut(BaseModel):
    """Единообразный формат ошибок API."""
    error: str
    detail: str | None = None


class MessageOut(BaseModel):
    """Простое сообщение об успехе."""
    message: str


class QuestionStatsOut(BaseModel):
    """Статистика голосов по одному вопросу."""
    model_config = ConfigDict(from_attributes=True)

    question_id: int
    question_text: str
    agree_count: int
    disagree_count: int
    total_count: int

    @property
    def agree_percent(self) -> float:
        """Процент голосов «за»."""
        if self.total_count == 0:
            return 0.0
        return round(self.agree_count / self.total_count * 100, 1)

2. Создаём blueprint статистики

# app/api/stats.py
from flask import Blueprint, jsonify
from sqlalchemy import func, select

from app import db
from app.models import Question, Response
from app.schemas.common import ErrorOut, QuestionStatsOut

stats_bp = Blueprint("stats", __name__, url_prefix="/api/stats")


def _build_stats_query():
    """
    Строит SQL-запрос для агрегации статистики.

    SQL-эквивалент:
        SELECT
            q.id,
            q.text,
            COUNT(CASE WHEN r.is_agree = TRUE THEN 1 END) AS agree_count,
            COUNT(CASE WHEN r.is_agree = FALSE THEN 1 END) AS disagree_count,
            COUNT(r.id) AS total_count
        FROM questions q
        LEFT JOIN responses r ON r.question_id = q.id
        GROUP BY q.id, q.text
        ORDER BY q.created_at DESC
    """
    agree_count = func.count(
        func.nullif(Response.is_agree == False, True)  # noqa: E712
    ).label("agree_count")
    disagree_count = func.count(
        func.nullif(Response.is_agree == True, True)   # noqa: E712
    ).label("disagree_count")
    total_count = func.count(Response.id).label("total_count")

    stmt = (
        select(
            Question.id.label("question_id"),
            Question.text.label("question_text"),
            agree_count,
            disagree_count,
            total_count,
        )
        .outerjoin(Response, Response.question_id == Question.id)
        .group_by(Question.id, Question.text)
        .order_by(Question.created_at.desc())
    )
    return stmt


@stats_bp.route("/questions/", methods=["GET"])
def get_all_stats():
    """
    GET /api/stats/questions/ — статистика по всем вопросам.

    Returns:
        200: список объектов с полями question_id, question_text,
             agree_count, disagree_count, total_count
    """
    rows = db.session.execute(_build_stats_query()).all()

    result = [
        {
            "question_id": row.question_id,
            "question_text": row.question_text,
            "agree_count": row.agree_count,
            "disagree_count": row.disagree_count,
            "total_count": row.total_count,
            "agree_percent": (
                round(row.agree_count / row.total_count * 100, 1)
                if row.total_count > 0 else 0.0
            ),
        }
        for row in rows
    ]

    return jsonify(result), 200


@stats_bp.route("/questions/<int:question_id>", methods=["GET"])
def get_question_stats(question_id: int):
    """GET /api/stats/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

    stmt = _build_stats_query().where(Question.id == question_id)
    row = db.session.execute(stmt).first()

    if not row:
        return jsonify({"question_id": question_id, "question_text": question.text,
                        "agree_count": 0, "disagree_count": 0, "total_count": 0,
                        "agree_percent": 0.0}), 200

    return jsonify({
        "question_id": row.question_id,
        "question_text": row.question_text,
        "agree_count": row.agree_count,
        "disagree_count": row.disagree_count,
        "total_count": row.total_count,
        "agree_percent": (
            round(row.agree_count / row.total_count * 100, 1)
            if row.total_count > 0 else 0.0
        ),
    }), 200

3. Централизованные error handlers

# app/errors.py
from flask import Flask, jsonify
from werkzeug.exceptions import HTTPException

from app.schemas.common import ErrorOut


def register_error_handlers(app: Flask) -> None:
    """
    Регистрирует глобальные обработчики ошибок.

    После регистрации:
        abort(404)  → {"error": "Not found", "detail": "..."}  с кодом 404
        abort(422)  → {"error": "Unprocessable entity", ...}   с кодом 422
        abort(500)  → {"error": "Internal server error", ...}  с кодом 500

    Любое необработанное исключение также попадёт в handler_500.
    """

    @app.errorhandler(400)
    def handler_400(error):
        return jsonify(ErrorOut(
            error="Bad request",
            detail=str(error.description) if hasattr(error, "description") else None,
        ).model_dump()), 400

    @app.errorhandler(404)
    def handler_404(error):
        return jsonify(ErrorOut(
            error="Not found",
            detail=str(error.description) if hasattr(error, "description") else None,
        ).model_dump()), 404

    @app.errorhandler(405)
    def handler_405(error):
        return jsonify(ErrorOut(
            error="Method not allowed",
            detail=str(error.description) if hasattr(error, "description") else None,
        ).model_dump()), 405

    @app.errorhandler(422)
    def handler_422(error):
        return jsonify(ErrorOut(
            error="Unprocessable entity",
            detail=str(error.description) if hasattr(error, "description") else None,
        ).model_dump()), 422

    @app.errorhandler(500)
    def handler_500(error):
        # В production не показываем внутренние детали клиенту
        app.logger.exception("Internal server error: %s", error)
        return jsonify(ErrorOut(
            error="Internal server error",
            detail="An unexpected error occurred. Please try again later.",
        ).model_dump()), 500

    @app.errorhandler(Exception)
    def handler_unhandled(error):
        """Поймать любое необработанное исключение."""
        if isinstance(error, HTTPException):
            # HTTPException обрабатывается выше
            return jsonify(ErrorOut(
                error=error.name,
                detail=str(error.description),
            ).model_dump()), error.code

        app.logger.exception("Unhandled exception: %s", error)
        return jsonify(ErrorOut(
            error="Internal server error",
            detail="An unexpected error occurred.",
        ).model_dump()), 500

4. Обновляем register_blueprints

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


def register_blueprints(app: Flask) -> None:
    """Регистрирует все API-blueprints."""
    from app.api.questions import questions_bp
    from app.api.responses import responses_bp
    from app.api.stats import stats_bp

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

5. Подключаем error handlers в 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)

    with app.app_context():
        from app.models import Question, Response  # noqa: F401

    from app.api import register_blueprints
    register_blueprints(app)

    # Централизованная обработка ошибок
    from app.errors import register_error_handlers
    register_error_handlers(app)

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

    return app

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

GROUP BY + COUNT: как работает агрегация

Запрос использует LEFT JOIN и GROUP BY для подсчёта голосов за один проход по БД. Без GROUP BY нам пришлось бы делать N запросов — по одному на каждый вопрос. Агрегация на уровне БД эффективнее: вместо N+1 запросов — один.

Error handlers vs явные проверки

До шага 06 каждый эндпоинт вручную проверял 404 и возвращал ErrorOut. Теперь можно использовать abort(404) — Flask вызовет handler. В следующих шагах можно упростить эндпоинты: заменить ручные проверки if not question: return jsonify(...), 404 на question = db.get_or_404(Question, id) (Flask-SQLAlchemy shorthand).

✅ Проверка

1. Тестируем статистику

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

curl -X POST http://localhost:5000/api/questions/1/responses/ \
  -H "Content-Type: application/json" \
  -d "{\"is_agree\": true}"

curl -X POST http://localhost:5000/api/questions/1/responses/ \
  -H "Content-Type: application/json" \
  -d "{\"is_agree\": false}"

# Получаем статистику
curl http://localhost:5000/api/stats/questions/
curl http://localhost:5000/api/stats/questions/1

2. Тестируем error handler

# Несуществующий вопрос — должен вернуть JSON-ошибку, не HTML
curl http://localhost:5000/api/questions/99999

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

[{
  "question_id": 1,
  "question_text": "Согласны ли вы с четырёхдневной рабочей неделей?",
  "agree_count": 1,
  "disagree_count": 1,
  "total_count": 2,
  "agree_percent": 50.0
}]

{"error": "Not found", "detail": "Question 99999 not found"}

➡️ Что дальше

На последнем шаге настроим логирование через Python logging, добавим prod-конфиг с переменными окружения, прогоним весь API через Postman и пройдём финальный чеклист production-ready проекта.

• Готово: агрегированная статистика, централизованные error handlers, единый формат ошибок
• Далее (шаг 07): logging, .env для prod, Postman-коллекция, финальный чеклист