Создаем Web API с Flask и SQLAlchemy

from flask import Flask, jsonify
 
app = Flask(__name__)
 
@app.route('/api/health-check', methods=['GET'])
def health_check():
    return jsonify({"status": "healthy", "version": "1.0.0"})
 
if __name__ == '__main__':
    app.run()

from flask_sqlalchemy import SQLAlchemy
 
db = SQLAlchemy()
 
class User(db.Model):
    id = db.Column(db.Integer, primary_key=True)
    username = db.Column(db.String(64), unique=True, nullable=False)
    email = db.Column(db.String(120), unique=True, nullable=False)
    
    def to_dict(self):
        return {
            'id': self.id,
            'username': self.username,
            'email': self.email
        }

# Flask - минималистично и понятно
@app.route('/users/<int:user_id>')
def get_user(user_id):
    user = User.query.get_or_404(user_id)
    return jsonify(user.to_dict())
 
# FastAPI - современно, но уже больше "магии"
@app.get('/users/{user_id}', response_model=UserResponse)
async def get_user(user_id: int):
    user = await find_user(user_id)
    if not user:
        raise HTTPException(status_code=404, detail="User not found")
    return user

from flask import Flask
from flask_restx import Api, Resource
from flask_sqlalchemy import SQLAlchemy
from flask_migrate import Migrate
from flask_jwt_extended import JWTManager
 
app = Flask(__name__)
api = Api(app, version='1.0', title='Todo API', description='A simple todo API')
db = SQLAlchemy(app)
migrate = Migrate(app, db)
jwt = JWTManager(app)
 
# И вот теперь у нас есть полнофункциональный бэкенд
# с документацией, ORM, миграциями и JWT-аутентификацией

import unittest
from app import create_app, db
 
class UserAPITestCase(unittest.TestCase):
    def setUp(self):
        self.app = create_app('testing')
        self.client = self.app.test_client()
        self.app_context = self.app.app_context()
        self.app_context.push()
        db.create_all()
        
    def tearDown(self):
        db.session.remove()
        db.drop_all()
        self.app_context.pop()
        
    def test_get_user(self):
        # Создаем тестового пользователя
        user = User(username='testuser', email='test@example.com')
        user.set_password('password')
        db.session.add(user)
        db.session.commit()
        
        # Проверяем получение
        response = self.client.get('/api/users/1')
        self.assertEqual(response.status_code, 200)
        json_response = response.get_json()
        self.assertEqual(json_response['username'], 'testuser')

import pytest
from app import create_app, db
from app.models import User
 
@pytest.fixture
def client():
    app = create_app('testing')
    with app.test_client() as client:
        with app.app_context():
            db.create_all()
            yield client
            db.session.remove()
            db.drop_all()
 
def test_get_user(client):
    # Создаем тестового пользователя
    user = User(username='testuser', email='test@example.com')
    user.set_password('password')
    db.session.add(user)
    db.session.commit()
    
    # Проверяем получение
    response = client.get('/api/users/1')
    assert response.status_code == 200
    json_response = response.get_json()
    assert json_response['username'] == 'testuser'

flask_api/
├── app/
│   ├── __init__.py  # Фабрика приложения
│   ├── config.py    # Конфигурация
│   ├── models/      # Модели данных
│   │   ├── __init__.py
│   │   ├── user.py
│   │   └── item.py
│   ├── routes/      # Маршруты API
│   │   ├── __init__.py
│   │   ├── user_routes.py
│   │   └── item_routes.py
│   ├── services/    # Бизнес-логика
│   │   ├── __init__.py
│   │   ├── user_service.py
│   │   └── item_service.py
│   └── utils/       # Утилиты
│       ├── __init__.py
│       ├── decorators.py
│       └── validators.py
├── migrations/      # Миграции БД
├── tests/           # Тесты
├── .env             # Переменные окружения
├── Dockerfile       # Контейнеризация
└── run.py          # Точка входа

from flask import Blueprint, jsonify, request
from app.services.user_service import UserService
 
user_bp = Blueprint('users', __name__, url_prefix='/api/users')
user_service = UserService()
 
@user_bp.route('/', methods=['GET'])
def get_users():
page = request.args.get('page', 1, type=int)
per_page = request.args.get('per_page', 20, type=int)
return jsonify(user_service.get_all_users(page, per_page))
 
@user_bp.route('/<int:user_id>', methods=['GET'])
def get_user(user_id):
user = user_service.get_user_by_id(user_id)
if not user:
    return jsonify({'error': 'User not found'}), 404
return jsonify(user)
 
@user_bp.route('/', methods=['POST'])
def create_user():
data = request.get_json()
user = user_service.create_user(data)
return jsonify(user), 201

from app.models.user import User
from app.models import db
 
class UserService:
def get_all_users(self, page=1, per_page=20):
    users = User.query.paginate(page, per_page, False)
    return {
        'items': [user.to_dict() for user in users.items],
        'total': users.total,
        'page': page,
        'per_page': per_page
    }
    
def get_user_by_id(self, user_id):
    user = User.query.get(user_id)
    return user.to_dict() if user else None
    
def create_user(self, data):
    user = User(
        username=data.get('username'),
        email=data.get('email')
    )
    user.set_password(data.get('password'))
    
    db.session.add(user)
    db.session.commit()
    
    return user.to_dict()

from flask import Flask
from flask_restx import Api, Resource, fields
 
app = Flask(__name__)
api = Api(app, version='1.0', title='User API',
    description='A simple User API')
 
ns = api.namespace('users', description='User operations')
 
user_model = api.model('User', {
'id': fields.Integer(readonly=True, description='Unique identifier'),
'username': fields.String(required=True, description='Username'),
'email': fields.String(required=True, description='Email address')
})
 
@ns.route('/')
class UserList(Resource):
@ns.doc('list_users')
@ns.marshal_list_with(user_model)
def get(self):
    """List all users"""
    return User.query.all()
 
@ns.route('/<int:id>')
@ns.response(404, 'User not found')
@ns.param('id', 'The user identifier')
class UserResource(Resource):
@ns.doc('get_user')
@ns.marshal_with(user_model)
def get(self, id):
    """Fetch a user by ID"""
    return User.query.get_or_404(id)

# Получение пользователя по ID
user = User.query.get(1)
 
# Фильтрация по условию
active_users = User.query.filter(User.is_active == True).all()
 
# Создание новой записи
new_item = Item(name="Новый товар", price=100)
db.session.add(new_item)
db.session.commit()

# Импорт необходимых компонентов
from sqlalchemy import select, text
 
# Чистый SQL через text()
raw_users = db.session.execute(text("SELECT * FROM users WHERE is_active = :active"), 
                              {"active": True}).fetchall()
 
# Использование построителя запросов
stmt = select(User).where(User.is_active == True)
core_users = db.session.execute(stmt).scalars().all()

class User(db.Model):
id = db.Column(db.Integer, primary_key=True)
username = db.Column(db.String(64), unique=True, nullable=False)
email = db.Column(db.String(120), unique=True, nullable=False)
posts = db.relationship('Post', backref='author', lazy='dynamic', 
                      cascade="all, delete-orphan")
 
class Post(db.Model):
id = db.Column(db.Integer, primary_key=True)
title = db.Column(db.String(128), nullable=False)
body = db.Column(db.Text)
created_at = db.Column(db.DateTime, default=datetime.utcnow)
user_id = db.Column(db.Integer, db.ForeignKey('user.id'), nullable=False)

class UserRepository:
def __init__(self, session):
    self.session = session
    
def get_by_id(self, user_id):
    return self.session.query(User).get(user_id)
    
def get_by_username(self, username):
    return self.session.query(User).filter(User.username == username).first()
    
def add(self, user):
    self.session.add(user)
    return user
    
def remove(self, user):
    self.session.delete(user)

class UserService:
def __init__(self, user_repository):
self.user_repository = user_repository
 
def register_user(self, username, email, password):
    # Проверяем, что пользователя с таким именем не существует
    if self.user_repository.get_by_username(username):
        raise ValueError("Username already exists")
        
    user = User(username=username, email=email)
    user.set_password(password)
    self.user_repository.add(user)
    return user

class UnitOfWork:
def __init__(self, session_factory):
    self.session_factory = session_factory
    
def __enter__(self):
    self.session = self.session_factory()
    self.users = UserRepository(self.session)
    self.posts = PostRepository(self.session)
    return self
    
def __exit__(self, exc_type, exc_val, exc_tb):
    self.session.close()
    
def commit(self):
    self.session.commit()
    
def rollback(self):
    self.session.rollback()

def register_user(username, email, password):
with UnitOfWork(session_factory) as uow:
    try:
        # Проверяем уникальность имени
        if uow.users.get_by_username(username):
            raise ValueError("Username already exists")
            
        user = User(username=username, email=email)
        user.set_password(password)
        uow.users.add(user)
        
        # Создаем приветственный пост
        welcome_post = Post(
            title="Welcome!",
            body="Welcome to our platform!",
            user_id=user.id
        )
        uow.posts.add(welcome_post)
        
        # Фиксируем все изменения
        uow.commit()
        return user
    except Exception as e:
        uow.rollback()
        raise e

# Антипаттерн: приводит к N+1 запросам
users = User.query.all()  # 1 запрос для получения всех пользователей
for user in users:
print(user.posts)  # N дополнительных запросов, по одному на каждого пользователя

# Оптимизировано: всего 1 запрос с JOIN
users = User.query.options(joinedload(User.posts)).all()
for user in users:
print(user.posts)  # Данные уже загружены, дополнительных запросов нет

# Выбираем только нужные поля
user_list = db.session.query(User.id, User.username).all()

report_data = db.session.execute("""
SELECT 
u.department_id, 
d.name as department_name,
COUNT(u.id) as user_count,
AVG(u.salary) as avg_salary
FROM users u
JOIN departments d ON u.department_id = d.id
GROUP BY u.department_id, d.name
ORDER BY user_count DESC
""").fetchall()

class User(db.Model):
id = db.Column(db.Integer, primary_key=True)
email = db.Column(db.String(120), index=True, unique=True)
department_id = db.Column(db.Integer, db.ForeignKey('department.id'), index=True)

app.config['SQLALCHEMY_ECHO'] = True  # Включить в режиме разработки!

# Создание сессии
session = db.session
 
try:
    # Выполнение операций
    user = User(username='new_user', email='new@example.com')
    session.add(user)
    
    # Еще операции...
    post = Post(title='First Post', user_id=user.id)
    session.add(post)
    
    # Фиксация изменений
    session.commit()
except Exception as e:
    # Отмена при ошибке
    session.rollback()
    raise e

# Внешняя транзакция
with session.begin():
    # Основные операции
    user = User(username='main_user', email='main@example.com')
    session.add(user)
    
    # Вложенная транзакция (savepoint)
    with session.begin_nested():
        try:
            # Рискованные операции
            risky_user = User(username='risky', email='risky@example.com')
            session.add(risky_user)
            # Что-то пошло не так...
            if some_condition:
                raise ValueError("Oops!")
        except Exception:
            # Внутренняя транзакция откатывается, но внешняя остается активной
            pass
            
    # Продолжение внешней транзакции
    post = Post(title='Safe Post', user_id=user.id)
    session.add(post)
    # Внешняя транзакция завершается успешно, даже если внутренняя откатилась

from flask_jwt_extended import create_access_token, jwt_required, get_jwt_identity
 
# Генерация токена при логине
@auth_bp.route('/login', methods=['POST'])
def login():
data = request.get_json()
user = User.query.filter_by(username=data.get('username')).first()
 
if not user or not user.check_password(data.get('password')):
    return jsonify({"error": "Invalid credentials"}), 401
    
# Создаем JWT токен с идентификатором пользователя
access_token = create_access_token(identity=user.id)
return jsonify(access_token=access_token), 200
 
# Защищенный маршрут, требующий валидный токен
@user_bp.route('/profile', methods=['GET'])
@jwt_required()
def profile():
# Извлекаем идентификатор пользователя из токена
current_user_id = get_jwt_identity()
user = User.query.get(current_user_id)
return jsonify(user.to_dict()), 200

class Role(db.Model):
id = db.Column(db.Integer, primary_key=True)
name = db.Column(db.String(80), unique=True)
permissions = db.relationship('Permission', secondary='role_permissions')
 
class Permission(db.Model):
id = db.Column(db.Integer, primary_key=True)
name = db.Column(db.String(80), unique=True)
 
# Таблица связи роль-разрешение
role_permissions = db.Table('role_permissions',
db.Column('role_id', db.Integer, db.ForeignKey('role.id')),
db.Column('permission_id', db.Integer, db.ForeignKey('permission.id'))
)
 
# Добавляем роли пользователю
user_roles = db.Table('user_roles',
db.Column('user_id', db.Integer, db.ForeignKey('user.id')),
db.Column('role_id', db.Integer, db.ForeignKey('role.id'))
)
 
# И дополняем модель User
class User(db.Model):
# ... (существующие поля)
roles = db.relationship('Role', secondary='user_roles')
 
def has_permission(self, permission_name):
    for role in self.roles:
        for permission in role.permissions:
            if permission.name == permission_name:
                return True
    return False

def permission_required(permission_name):
def decorator(f):
    @wraps(f)
    @jwt_required()
    def decorated_function(*args, **kwargs):
        current_user_id = get_jwt_identity()
        user = User.query.get(current_user_id)
        
        if not user.has_permission(permission_name):
            return jsonify({"error": "Permission denied"}), 403
            
        return f(*args, **kwargs)
    return decorated_function
return decorator
 
# Применение
@user_bp.route('/admin/users', methods=['GET'])
@permission_required('view_all_users')
def get_all_users():
# Только пользователи с правом 'view_all_users' могут вызвать этот метод
users = User.query.all()
return jsonify([user.to_dict() for user in users]), 200

from marshmallow import Schema, fields, validate, ValidationError
 
class UserSchema(Schema):
username = fields.String(required=True, validate=validate.Length(min=3, max=50))
email = fields.Email(required=True)
age = fields.Integer(validate=validate.Range(min=18))
 
@user_bp.route('/', methods=['POST'])
def create_user():
schema = UserSchema()
try:
    # Валидация и десериализация
    data = schema.load(request.get_json())
except ValidationError as err:
    return jsonify({"errors": err.messages}), 400
    
# Теперь data содержит только валидные данные
user = User(**data)
db.session.add(user)
db.session.commit()
return jsonify(schema.dump(user)), 201

@app.before_request
def validate_api_key():
# Пропускаем открытые маршруты
if request.endpoint in ['public.health', 'auth.login']:
return
 
# Проверяем ключ API в заголовках
api_key = request.headers.get('X-API-Key')
if not api_key:
    return jsonify({"error": "API key missing"}), 401
    
# Проверяем валидность ключа
api_key_record = ApiKey.query.filter_by(key=api_key).first()
if not api_key_record or not api_key_record.is_active:
    return jsonify({"error": "Invalid or inactive API key"}), 401
    
# Добавляем информацию о владельце ключа в глобальный контекст
g.api_client = api_key_record.client

from flask_limiter import Limiter
from flask_limiter.util import get_remote_address
 
limiter = Limiter(
app,
key_func=get_remote_address,  # Идентификация по IP
default_limits=["200 per day", "50 per hour"]  # Глобальные ограничения
)
 
# Применение к конкретному маршруту
@user_bp.route('/reset-password', methods=['POST'])
@limiter.limit("3 per hour")  # Ограничение на сброс пароля
def reset_password():
# Логика сброса пароля

from authlib.integrations.flask_client import OAuth
 
oauth = OAuth(app)
github = oauth.register(
'github',
client_id='YOUR_CLIENT_ID',
client_secret='YOUR_CLIENT_SECRET',
access_token_url='https://github.com/login/oauth/access_token',
access_token_params=None,
authorize_url='https://github.com/login/oauth/authorize',
authorize_params=None,
api_base_url='https://api.github.com/',
client_kwargs={'scope': 'user:email'},
)
 
@auth_bp.route('/login/github')
def github_login():
redirect_uri = url_for('auth.github_authorize', _external=True)
return github.authorize_redirect(redirect_uri)
 
@auth_bp.route('/authorize/github')
def github_authorize():
token = github.authorize_access_token()
resp = github.get('user')
profile = resp.json()
# Логика создания/обновления пользователя и входа в систему

from flask_caching import Cache
 
cache = Cache(config={'CACHE_TYPE': 'simple'})  # Для разработки
[H2]cache = Cache(config={'CACHE_TYPE': 'redis', 'CACHE_REDIS_URL': 'redis://localhost:6379/0'})  # Для прода[/H2]
 
def create_app():
app = Flask(__name__)
cache.init_app(app)
return app
 
@app.route('/api/products')
@cache.cached(timeout=60)  # Кешируем на 60 секунд
def get_products():
# Тяжелый запрос к БД
products = Product.query.all()
return jsonify([p.to_dict() for p in products])

@cache.memoize(timeout=300)  # Кешируем результаты для каждого набора параметров
def get_user_stats(user_id):
# Сложные вычисления или тяжелые запросы
return StatisticsService.calculate_user_stats(user_id)
 
@app.route('/api/users/<int:user_id>/stats')
def user_stats(user_id):
return jsonify(get_user_stats(user_id))
 
# Можно также инвалидировать кеш для конкретного пользователя
@app.route('/api/users/<int:user_id>/stats/refresh', methods=['POST'])
def refresh_user_stats(user_id):
cache.delete_memoized(get_user_stats, user_id)
return jsonify(get_user_stats(user_id))  # Свежие данные

from flask import Flask
from celery import Celery
 
app = Flask(__name__)
app.config.update(
CELERY_BROKER_URL='redis://localhost:6379/1',
CELERY_RESULT_BACKEND='redis://localhost:6379/1'
)
 
celery = Celery(app.name, broker=app.config['CELERY_BROKER_URL'])
celery.conf.update(app.config)
 
@celery.task
def process_data(data):
# Долгая обработка данных
result = some_heavy_processing(data)
return result
 
@app.route('/api/process', methods=['POST'])
def submit_processing():
data = request.get_json()
task = process_data.delay(data)
return jsonify({'task_id': task.id}), 202  # Accepted
 
@app.route('/api/tasks/<task_id>')
def get_task_result(task_id):
task = process_data.AsyncResult(task_id)
if task.state == 'PENDING':
return jsonify({'status': 'pending'})
elif task.state == 'SUCCESS':
return jsonify({'status': 'completed', 'result': task.result})
else:
return jsonify({'status': 'failed', 'error': str(task.result)})

# Простое профилирование времени выполнения
import time
from functools import wraps
 
def timing_decorator(f):
@wraps(f)
def wrapper(*args, **kwargs):
    start = time.time()
    result = f(*args, **kwargs)
    end = time.time()
    print(f"{f.__name__} took {end - start:.2f} seconds")
    return result
return wrapper
 
@app.route('/api/expensive-operation')
@timing_decorator
def expensive_operation():
# Какие-то сложные вычисления
return result

# Плохо: хранение состояния в памяти приложения
user_data = {}  # Глобальный словарь для хранения данных
 
@app.route('/api/set_preference')
def set_preference():
    user_id = request.args.get('user_id')
    preference = request.args.get('preference')
    user_data[user_id] = preference
    return jsonify({'status': 'ok'})
 
# Хорошо: все состояния хранятся во внешнем хранилище
@app.route('/api/set_preference')
def set_preference():
    user_id = request.args.get('user_id')
    preference = request.args.get('preference')
    redis_client.set(f"user:{user_id}:preference", preference)
    return jsonify({'status': 'ok'})

from flask import Flask
from flask_session import Session
from redis import Redis
 
app = Flask(__name__)
app.config.update(
    SESSION_TYPE='redis',
    SESSION_REDIS=Redis(host='redis_server', port=6379)
)
Session(app)

from flask_sqlalchemy import SQLAlchemy
 
app = Flask(__name__)
app.config['SQLALCHEMY_DATABASE_URI'] = 'postgresql://user:pass@postgres/dbname'
app.config['SQLALCHEMY_ENGINE_OPTIONS'] = {
    'pool_size': 10,
    'max_overflow': 20,
    'pool_recycle': 300,  # Переиспользовать соединения после 5 минут
    'pool_pre_ping': True,  # Проверять соединения перед использованием
}
db = SQLAlchemy(app)

import logging
import json
from pythonjsonlogger import jsonlogger
 
class CustomJsonFormatter(jsonlogger.JsonFormatter):
    def add_fields(self, log_record, record, message_dict):
        super(CustomJsonFormatter, self).add_fields(log_record, record, message_dict)
        log_record['service'] = 'user-api'
        log_record['hostname'] = socket.gethostname()
 
handler = logging.StreamHandler()
formatter = CustomJsonFormatter('%(timestamp)s %(level)s %(name)s %(message)s')
handler.setFormatter(formatter)
 
logger = logging.getLogger('flask_api')
logger.addHandler(handler)
logger.setLevel(logging.INFO)
 
# Использование в коде
@app.route('/api/users/<int:user_id>')
def get_user(user_id):
    logger.info('User requested', extra={
        'user_id': user_id,
        'ip': request.remote_addr,
        'endpoint': request.endpoint
    })
    # ...остальной код

from prometheus_client import Counter, Histogram
from prometheus_flask_exporter import PrometheusMetrics
 
metrics = PrometheusMetrics(app)
 
# Счетчик запросов по статус-кодам
status_counter = Counter(
    'api_requests_total',
    'Total count of API requests by status code',
    ['status', 'endpoint']
)
 
# Гистограмма времени ответа
response_time = Histogram(
    'api_response_time_seconds',
    'Response time in seconds',
    ['endpoint']
)
 
@app.after_request
def track_request(response):
    status_counter.labels(
        status=response.status_code,
        endpoint=request.endpoint or 'unknown'
    ).inc()
    return response
 
@app.route('/api/slow-operation')
@response_time.labels(endpoint='/api/slow-operation').time()
def slow_operation():
    # Ваш код здесь
    return result

app/
  __init__.py
  routes.py
  models.py
config.py
run.py

app/
  __init__.py
  models/
  routes/
  services/
  schemas/
  utils/
migrations/
tests/

# Пример для гибридного подхода: точка входа в API-шлюз
from flask import Flask, request, jsonify
import requests
 
app = Flask(__name__)
 
# Монолитная часть
@app.route('/api/users', methods=['GET'])
def get_users():
  # Логика получения пользователей
  return jsonify(users)
 
# Проксирование запроса к микросервису
@app.route('/api/analytics', methods=['GET'])
def get_analytics():
  response = requests.get('http://analytics-service/api/data', 
                        params=request.args)
  return jsonify(response.json())

# Типичная операция через ORM
user = User.query.get(user_id)
 
# Оптимизированный вариант для критического пути
from sqlalchemy import text
user_data = db.session.execute(
  text("""
  SELECT id, username, email 
  FROM users 
  WHERE id = :user_id 
  LIMIT 1
  """), 
  {"user_id": user_id}
).fetchone()

# Традиционный синхронный подход
@app.route('/api/data')
def get_data():
  # Блокирущий вызов внешнего API
  external_data = requests.get('https://external-api.com/data').json()
  # Обработка данных
  return jsonify(result)
 
# Асинхронный подход с Flask + Quart
@app.route('/api/data')
async def get_data():
  # Неблокирующий вызов
  async with httpx.AsyncClient() as client:
      external_data = await client.get('https://external-api.com/data')
  # Обработка данных
  return jsonify(result)

# Пример безопасной миграции - сначала создаем новую колонку
[H2]op.add_column('users', sa.Column('full_name', sa.String(120)))[/H2]
 
# Затем в коде приложения начинаем заполнять эту колонку
@event.listens_for(User, 'before_insert')
def set_full_name(mapper, connection, target):
  if target.first_name and target.last_name:
      target.full_name = f"{target.first_name} {target.last_name}"
 
# В следующей миграции можем безопасно удалить старые колонки
# op.drop_column('users', 'first_name')
# op.drop_column('users', 'last_name')