💻 Примеры: Community Pulse в одном месте

1. Конфигурация приложения

# 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:'

2. Application Factory

# app/__init__.py
from flask import Flask
from flask_migrate import Migrate
from app.models import db
from app.routers.questions import questions_bp
from app.routers.response import response_bp

def create_app(config=None):
    app = Flask(__name__)
    if config:
        app.config.from_object(config)
    db.init_app(app)
    Migrate(app, db)
    app.register_blueprint(questions_bp)
    app.register_blueprint(response_bp)
    return app

3. Модели данных

# app/models/__init__.py
from flask_sqlalchemy import SQLAlchemy
db = SQLAlchemy()
from app.models.questions import *
from app.models.response import *

# app/models/questions.py
from app.models import db

class Question(db.Model):
    __tablename__ = 'questions'
    id = db.Column(db.Integer, primary_key=True)
    text = db.Column(db.String(255), nullable=False)
    responses = db.relationship('Response', backref='question', lazy=True)

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

class Statistic(db.Model):
    __tablename__ = 'statistics'
    question_id = db.Column(db.Integer, db.ForeignKey('questions.id'), primary_key=True)
    agree_count = db.Column(db.Integer, nullable=False, default=0)
    disagree_count = db.Column(db.Integer, nullable=False, default=0)

# app/models/response.py
from app.models import db

class Response(db.Model):
    __tablename__ = 'responses'
    id = db.Column(db.Integer, primary_key=True)
    question_id = db.Column(db.Integer, db.ForeignKey('questions.id'), nullable=False)
    is_agree = db.Column(db.Boolean, nullable=False)

4. Pydantic-схемы

# app/schemas/question.py  (Pydantic v2)
from pydantic import BaseModel, Field, ConfigDict

class QuestionCreate(BaseModel):
    text: str = Field(..., min_length=12, description="Текст вопроса (мин. 12 символов)")

class QuestionResponse(BaseModel):
    model_config = ConfigDict(from_attributes=True)
    id: int
    text: str

class MessageResponse(BaseModel):
    message: str

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

class ResponseCreate(BaseModel):
    question_id: int = Field(..., description="ID вопроса")
    is_agree: bool = Field(..., description="Согласен (True) или нет (False)")

class StatisticResponse(BaseModel):
    model_config = ConfigDict(from_attributes=True)
    question_id: int
    agree_count: int
    disagree_count: int

5. Blueprint для вопросов (полный CRUD)

# app/routers/questions.py
from flask import Blueprint, request, jsonify
from pydantic import ValidationError
from app.models import db
from app.models.questions import Question
from app.schemas.question import QuestionCreate, QuestionResponse

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

@questions_bp.route('/', methods=['GET'])
def get_questions():
    questions = Question.query.all()
    results = [QuestionResponse.model_validate(q).model_dump() for q in questions]
    return jsonify(results), 200

@questions_bp.route('/', methods=['POST'])
def create_question():
    data = request.get_json()
    try:
        question_data = QuestionCreate(**data)
    except (ValidationError, TypeError) as e:
        return jsonify({'error': str(e)}), 400
    question = Question(text=question_data.text)
    db.session.add(question)
    db.session.commit()
    return jsonify(QuestionResponse.model_validate(question).model_dump()), 201

@questions_bp.route('/<int:id>', methods=['GET'])
def get_question(id):
    question = db.session.get(Question, id)
    if question is None:
        return jsonify({'message': 'Вопрос не найден'}), 404
    return jsonify(QuestionResponse.model_validate(question).model_dump()), 200

@questions_bp.route('/<int:id>', methods=['PUT'])
def update_question(id):
    question = db.session.get(Question, id)
    if question is None:
        return jsonify({'message': 'Вопрос не найден'}), 404
    data = request.get_json()
    if 'text' not in data:
        return jsonify({'message': 'Поле text обязательно'}), 400
    question.text = data['text']
    db.session.commit()
    return jsonify(QuestionResponse.model_validate(question).model_dump()), 200

@questions_bp.route('/<int:id>', methods=['DELETE'])
def delete_question(id):
    question = db.session.get(Question, id)
    if question is None:
        return jsonify({'message': 'Вопрос не найден'}), 404
    db.session.delete(question)
    db.session.commit()
    return jsonify({'message': f'Вопрос {id} удалён'}), 200

6. Blueprint для ответов

# app/routers/response.py
from flask import Blueprint, request, jsonify
from app.models import db
from app.models.questions import Question, Statistic
from app.models.response import Response
from app.schemas.response import ResponseCreate, StatisticResponse

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

@response_bp.route('/', methods=['GET'])
def get_responses():
    statistics = Statistic.query.all()
    results = [StatisticResponse.model_validate(s).model_dump() for s in statistics]
    return jsonify(results), 200

@response_bp.route('/', methods=['POST'])
def add_response():
    data = request.get_json()
    try:
        resp_data = ResponseCreate(**data)
    except Exception as e:
        return jsonify({'error': str(e)}), 400

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

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

    stat = Statistic.query.filter_by(question_id=resp_data.question_id).first()
    if not stat:
        stat = Statistic(question_id=resp_data.question_id, agree_count=0, disagree_count=0)
        db.session.add(stat)
    if resp_data.is_agree:
        stat.agree_count += 1
    else:
        stat.disagree_count += 1

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

7. Запуск приложения

# run.py
from app import create_app
from config import DevelopmentConfig

app = create_app(DevelopmentConfig)

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