💻 Примеры: Community Pulse — полный код

Шаг 1: config.py

# config.py

class Config:
    DEBUG = False
    TESTING = False
    SQLALCHEMY_DATABASE_URI = 'sqlite:///community_pulse.db'
    SQLALCHEMY_TRACK_MODIFICATIONS = False

class DevelopmentConfig(Config):
    DEBUG = True

class TestingConfig(Config):
    TESTING = True
    SQLALCHEMY_DATABASE_URI = 'sqlite:///:memory:'

class ProductionConfig(Config):
    pass

config_map = {
    'development': DevelopmentConfig,
    'testing': TestingConfig,
    'production': ProductionConfig,
}

Шаг 2: Модели (app/models/questions.py)

# app/models/questions.py

from app import db

class Question(db.Model):
    __tablename__ = 'questions'

    id = db.Column(db.Integer, primary_key=True, autoincrement=True)
    text = db.Column(db.String(500), nullable=False)

    # связь с таблицей responses
    responses = db.relationship('Response', backref='question', lazy=True)

    def __repr__(self):
        return f'<Question {self.id}: {self.text[:30]}>'

Шаг 3: Модели (app/models/responses.py)

# app/models/responses.py

from app import db

class Response(db.Model):
    __tablename__ = 'responses'

    id = db.Column(db.Integer, primary_key=True, autoincrement=True)
    question_id = db.Column(
        db.Integer,
        db.ForeignKey('questions.id'),
        nullable=False
    )
    agree = db.Column(db.Boolean, nullable=False)  # True=согласен, False=не согласен

    def __repr__(self):
        status = "agree" if self.agree else "disagree"
        return f'<Response {self.id}: {status} on Q{self.question_id}>'

Шаг 4: Схемы Pydantic (app/schemas/question.py)

# app/schemas/question.py

from pydantic import BaseModel, Field

class QuestionCreate(BaseModel):
    """Схема для создания вопроса (входные данные)."""
    text: str = Field(..., min_length=5, max_length=500, description="Текст вопроса")

class QuestionOut(BaseModel):
    """Схема для возврата вопроса (выходные данные)."""
    id: int
    text: str

    model_config = {"from_attributes": True}  # Pydantic v2: читает из ORM-объектов

Шаг 5: Blueprint questions (app/routers/questions.py)

# app/routers/questions.py

from flask import Blueprint, jsonify, request
from app import db
from app.models.questions import Question
from app.schemas.question import QuestionCreate, QuestionOut
from sqlalchemy import select
from pydantic import ValidationError

questions_bp = Blueprint('questions', __name__)

@questions_bp.route('/', methods=['GET'])
def list_questions():
    """GET /questions/ — список всех вопросов."""
    questions = db.session.scalars(select(Question)).all()
    result = [QuestionOut.model_validate(q).model_dump() for q in questions]
    return jsonify(result)

@questions_bp.route('/', methods=['POST'])
def create_question():
    """POST /questions/ — создать новый вопрос."""
    data = request.get_json()
    try:
        schema = QuestionCreate(**data)
    except ValidationError as e:
        return jsonify({"error": e.errors()}), 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):
    """GET /questions/<id> — вопрос по ID."""
    question = db.session.get(Question, question_id)
    if not question:
        return jsonify({"error": "Not found"}), 404
    return jsonify(QuestionOut.model_validate(question).model_dump())

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

    data = request.get_json()
    try:
        schema = QuestionCreate(**data)
    except ValidationError as e:
        return jsonify({"error": e.errors()}), 422

    question.text = schema.text
    db.session.commit()
    return jsonify(QuestionOut.model_validate(question).model_dump())

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

    db.session.delete(question)
    db.session.commit()
    return jsonify({"message": "Deleted"}), 200

Шаг 6: Application Factory (app/__init__.py)

# app/__init__.py

from flask import Flask
from flask_sqlalchemy import SQLAlchemy

db = SQLAlchemy()

def create_app(config_name='development'):
    app = Flask(__name__)

    from config import config_map
    app.config.from_object(config_map[config_name])

    db.init_app(app)

    from app.routers.questions import questions_bp
    from app.routers.responses import responses_bp
    app.register_blueprint(questions_bp, url_prefix='/questions')
    app.register_blueprint(responses_bp, url_prefix='/responses')

    # создать таблицы при первом запуске (только для разработки)
    with app.app_context():
        db.create_all()

    return app

Шаг 7: Точка входа run.py

# run.py

import os
from app import create_app

config_name = os.environ.get('APP_ENV', 'development')
app = create_app(config_name)

if __name__ == '__main__':
    app.run()

Шаг 8: Blueprint responses (app/routers/responses.py)

# app/routers/responses.py

from flask import Blueprint, jsonify, request
from app import db
from app.models.responses import Response
from app.models.questions import Question
from sqlalchemy import select

responses_bp = Blueprint('responses', __name__)

@responses_bp.route('/', methods=['GET'])
def list_responses():
    """GET /responses/ — статистика ответов."""
    responses = db.session.scalars(select(Response)).all()
    result = [
        {"id": r.id, "question_id": r.question_id, "agree": r.agree}
        for r in responses
    ]
    return jsonify(result)

@responses_bp.route('/', methods=['POST'])
def create_response():
    """POST /responses/ — добавить ответ на вопрос."""
    data = request.get_json()
    question_id = data.get('question_id')
    agree = data.get('agree')

    if question_id is None or agree is None:
        return jsonify({"error": "question_id and agree are required"}), 400

    question = db.session.get(Question, question_id)
    if not question:
        return jsonify({"error": "Question not found"}), 404

    response = Response(question_id=question_id, agree=bool(agree))
    db.session.add(response)
    db.session.commit()

    return jsonify({"id": response.id, "question_id": question_id, "agree": agree}), 201

Тестирование через curl

# Запуск
python run.py

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

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

# Ответить на вопрос
curl -X POST http://localhost:5000/responses/ \
     -H "Content-Type: application/json" \
     -d '{"question_id": 1, "agree": true}'

# Статистика
curl http://localhost:5000/responses/