Делал ли какие-то интеграции с базой данной на Flask

Интеграции с базой данных на Flask

Да, я имею значительный опыт работы с интеграциями баз данных в Flask приложениях. Это ключевой компонент практически любого веб-приложения, и я хотел бы поделиться моим опытом и лучшими практиками.

Основные подходы к интеграции БД в Flask

SQLAlchemy ORM Это наиболее популярный и рекомендуемый подход. Flask-SQLAlchemy расширение предоставляет удобный интерфейс для работы с различными РСУБД (PostgreSQL, MySQL, SQLite).

from flask import Flask
from flask_sqlalchemy import SQLAlchemy
from datetime import datetime

app = Flask(__name__)
app.config['SQLALCHEMY_DATABASE_URI'] = 'postgresql://user:password@localhost/mydb'
db = SQLAlchemy(app)

class User(db.Model):
    __tablename__ = 'users'
    
    id = db.Column(db.Integer, primary_key=True)
    email = db.Column(db.String(120), unique=True, nullable=False, index=True)
    username = db.Column(db.String(80), nullable=False)
    password_hash = db.Column(db.String(255), nullable=False)
    created_at = db.Column(db.DateTime, default=datetime.utcnow)
    
    posts = db.relationship('Post', backref='author', lazy='select', cascade='all, delete-orphan')
    
    def __repr__(self):
        return f'<User {self.username}>'

class Post(db.Model):
    __tablename__ = 'posts'
    
    id = db.Column(db.Integer, primary_key=True)
    title = db.Column(db.String(200), nullable=False)
    content = db.Column(db.Text)
    user_id = db.Column(db.Integer, db.ForeignKey('users.id'), nullable=False)
    created_at = db.Column(db.DateTime, default=datetime.utcnow)

# Создание таблиц
with app.app_context():
    db.create_all()

Операции CRUD

Create (Создание)

@app.route('/users', methods=['POST'])
def create_user():
    data = request.get_json()
    
    user = User(
        email=data['email'],
        username=data['username'],
        password_hash=hash_password(data['password'])
    )
    
    db.session.add(user)
    db.session.commit()
    
    return jsonify({'id': user.id, 'email': user.email}), 201

Read (Чтение)

@app.route('/users/<int:user_id>', methods=['GET'])
def get_user(user_id):
    user = User.query.get_or_404(user_id)
    return jsonify({
        'id': user.id,
        'email': user.email,
        'username': user.username,
        'created_at': user.created_at.isoformat()
    })

@app.route('/users', methods=['GET'])
def list_users():
    page = request.args.get('page', 1, type=int)
    per_page = 20
    
    paginated = User.query.paginate(page=page, per_page=per_page)
    
    return jsonify({
        'items': [{'id': u.id, 'email': u.email} for u in paginated.items],
        'total': paginated.total,
        'pages': paginated.pages
    })

Update (Обновление)

@app.route('/users/<int:user_id>', methods=['PUT'])
def update_user(user_id):
    user = User.query.get_or_404(user_id)
    data = request.get_json()
    
    if 'email' in data:
        user.email = data['email']
    if 'username' in data:
        user.username = data['username']
    
    db.session.commit()
    return jsonify({'message': 'User updated'})

Delete (Удаление)

@app.route('/users/<int:user_id>', methods=['DELETE'])
def delete_user(user_id):
    user = User.query.get_or_404(user_id)
    db.session.delete(user)
    db.session.commit()
    return jsonify({'message': 'User deleted'}), 204

Сложные запросы

Фильтрация и сортировка

@app.route('/posts', methods=['GET'])
def list_posts():
    author_id = request.args.get('author_id', type=int)
    sort_by = request.args.get('sort', 'created_at')
    
    query = Post.query
    
    if author_id:
        query = query.filter_by(user_id=author_id)
    
    posts = query.order_by(sort_by).all()
    
    return jsonify([
        {'id': p.id, 'title': p.title, 'author': p.author.username}
        for p in posts
    ])

Join запросы

@app.route('/users-with-posts', methods=['GET'])
def users_with_posts():
    results = db.session.query(User, db.func.count(Post.id)).join(Post).group_by(User.id).all()
    
    return jsonify([
        {'user': u.username, 'post_count': count}
        for u, count in results
    ])

Транзакции и обработка ошибок

@app.route('/transfer', methods=['POST'])
def transfer_funds():
    try:
        from_user_id = request.json['from']
        to_user_id = request.json['to']
        amount = request.json['amount']
        
        from_user = User.query.get_or_404(from_user_id)
        to_user = User.query.get_or_404(to_user_id)
        
        if from_user.balance < amount:
            return jsonify({'error': 'Insufficient funds'}), 400
        
        from_user.balance -= amount
        to_user.balance += amount
        
        db.session.commit()
        return jsonify({'message': 'Transfer successful'}), 200
        
    except Exception as e:
        db.session.rollback()
        return jsonify({'error': str(e)}), 500

Миграции БД с Alembic

# Инициализация
# flask db init
# flask db migrate -m "Initial migration"
# flask db upgrade

from flask_migrate import Migrate

migrate = Migrate(app, db)

Оптимизация производительности

Использование индексов

class User(db.Model):
    id = db.Column(db.Integer, primary_key=True)
    email = db.Column(db.String(120), unique=True, index=True)
    username = db.Column(db.String(80), index=True)

Eager loading для избежания N+1 проблемы

posts = Post.query.options(db.joinedload(Post.author)).all()

Кэширование результатов

from flask_caching import Cache

cache = Cache(app, config={'CACHE_TYPE': 'simple'})

@app.route('/users/<int:user_id>')
@cache.cached(timeout=300)
def get_user(user_id):
    return db.session.query(User).get(user_id)

Заключение

В моем опыте я работал с:

• PostgreSQL и MySQL интеграциями на Flask
• Сложными отношениями между сущностями (many-to-many, полиморфные связи)
• Миграциями БД и версионированием схемы
• Оптимизацией запросов и предотвращением N+1 проблемы
• Обработкой транзакций и откатом при ошибках

Это делает меня уверенным разработчиком при работе с персистентностью данных в Flask приложениях.