💻 Примеры: полный CRUD Community Pulse

Пример 1: Полный файл questions.py (SQLAlchemy 2.x)

# app/routers/questions.py
from flask import Blueprint, request, jsonify
from sqlalchemy import select
from app.models import db
from app.models.questions import Question

questions_bp = Blueprint('questions', __name__, url_prefix='/questions')


@questions_bp.route('/', methods=['GET'])
def get_questions():
    """Получение списка всех вопросов."""
    questions = db.session.execute(select(Question)).scalars().all()
    questions_data = [{'id': q.id, 'text': q.text} for q in questions]
    return jsonify(questions_data)


@questions_bp.route('/<int:id>', methods=['GET'])
def get_question(id):
    """Получение деталей конкретного вопроса по его ID."""
    question = db.session.get(Question, id)
    if question is None:
        return jsonify({'message': 'Вопрос с таким ID не найден'}), 404
    return jsonify({'id': question.id, 'text': question.text}), 200


@questions_bp.route('/', methods=['POST'])
def create_question():
    """Создание нового вопроса."""
    data = request.get_json()
    if not data or 'text' not in data:
        return jsonify({'error': 'No question text provided'}), 400

    question = Question(text=data['text'])
    db.session.add(question)
    db.session.commit()
    return jsonify({'message': 'Вопрос создан', 'id': question.id}), 201


@questions_bp.route('/<int:id>', methods=['PUT'])
def update_question(id):
    """Обновление конкретного вопроса по его ID."""
    question = db.session.get(Question, id)
    if question is None:
        return jsonify({'message': 'Вопрос с таким ID не найден'}), 404

    data = request.get_json()
    if not data or 'text' not in data:
        return jsonify({'message': 'Текст вопроса не предоставлен'}), 400

    question.text = data['text']
    db.session.commit()
    return jsonify({'message': f'Вопрос обновлён: {question.text}'}), 200


@questions_bp.route('/<int:id>', methods=['DELETE'])
def delete_question(id):
    """Удаление конкретного вопроса по его ID."""
    question = db.session.get(Question, id)
    if question is None:
        return jsonify({'message': 'Вопрос с таким ID не найден'}), 404

    db.session.delete(question)
    db.session.commit()
    return jsonify({'message': f'Вопрос с ID {id} удалён'}), 200

Пример 2: Полный файл response.py (статистика ответов)

# app/routers/response.py
from flask import Blueprint, request, jsonify
from sqlalchemy import select
from app.models import db
from app.models.questions import Question
from app.models.statistic import Statistic
from app.models.response import Response

response_bp = Blueprint('response', __name__, url_prefix='/responses')


@response_bp.route('/', methods=['GET'])
def get_responses():
    """Получение агрегированной статистики ответов."""
    statistics = db.session.execute(select(Statistic)).scalars().all()
    results = [
        {
            'question_id': stat.question_id,
            'agree_count': stat.agree_count,
            'disagree_count': stat.disagree_count
        }
        for stat in statistics
    ]
    return jsonify(results), 200


@response_bp.route('/', methods=['POST'])
def add_response():
    """Добавление нового ответа на вопрос с обновлением статистики."""
    data = request.get_json()
    if not data or 'question_id' not in data or 'is_agree' not in data:
        return jsonify({'message': 'Некорректные данные'}), 400

    question = db.session.get(Question, data['question_id'])
    if not question:
        return jsonify({'message': 'Вопрос не найден'}), 404

    # Создаём запись ответа
    response = Response(
        question_id=question.id,
        is_agree=data['is_agree']
    )
    db.session.add(response)

    # Обновляем или создаём статистику
    statistic = db.session.execute(
        select(Statistic).filter_by(question_id=question.id)
    ).scalar_one_or_none()

    if not statistic:
        statistic = Statistic(
            question_id=question.id,
            agree_count=0,
            disagree_count=0
        )
        db.session.add(statistic)

    if data['is_agree']:
        statistic.agree_count += 1
    else:
        statistic.disagree_count += 1

    db.session.commit()
    return jsonify({'message': f'Ответ на вопрос {question.id} добавлен'}), 201

Пример 3: Регистрация blueprints в create_app()

# app/__init__.py
from flask import Flask
from app.models import db

def create_app(config=None):
    app = Flask(__name__)
    app.config.from_object(config or 'app.config.DevelopmentConfig')

    db.init_app(app)

    from app.routers.questions import questions_bp
    from app.routers.response import response_bp

    app.register_blueprint(questions_bp)
    app.register_blueprint(response_bp)

    return app

Пример 4: Тестирование в Postman

GET /questions — получить все вопросы

Method: GET
URL: http://localhost:5000/questions/
Headers: (не нужны)
Expected: 200 OK, JSON-массив

POST /questions — создать вопрос

Method: POST
URL: http://localhost:5000/questions/
Headers: Content-Type: application/json
Body (raw JSON):
{
  "text": "Согласны ли вы с ограничением скорости 30 км/ч в городе?"
}
Expected: 201 Created
{
  "message": "Вопрос создан",
  "id": 1
}

PUT /questions/1 — обновить вопрос

Method: PUT
URL: http://localhost:5000/questions/1
Headers: Content-Type: application/json
Body (raw JSON):
{
  "text": "Обновлённый текст вопроса"
}
Expected: 200 OK

DELETE /questions/1 — удалить вопрос

Method: DELETE
URL: http://localhost:5000/questions/1
Expected: 200 OK
{
  "message": "Вопрос с ID 1 удалён"
}

POST /responses — добавить ответ

Method: POST
URL: http://localhost:5000/responses/
Headers: Content-Type: application/json
Body (raw JSON):
{
  "question_id": 1,
  "is_agree": true
}
Expected: 201 Created
{
  "message": "Ответ на вопрос 1 добавлен"
}

Пример 5: Использование request.args для фильтрации

@questions_bp.route('/', methods=['GET'])
def get_questions():
    """Получение вопросов с опциональным поиском."""
    search = request.args.get('search', '')  # /questions/?search=скорость
    stmt = select(Question)
    if search:
        stmt = stmt.where(Question.text.ilike(f'%{search}%'))
    questions = db.session.execute(stmt).scalars().all()
    return jsonify([{'id': q.id, 'text': q.text} for q in questions])