Объединение Flask с SQLAlchemy: практические варианты и их реализация

Расширенные примеры и нестандартные приёмы

Здесь приведены более сложные сценарии использования SQLAlchemy совместно с Flask.

1. Работа с отношениями (один-ко-многим)

Пример модели с внешним ключом и связью.

from flask_sqlalchemy import SQLAlchemy
db = SQLAlchemy()

class Author(db.Model):
    id = db.Column(db.Integer, primary_key=True)
    name = db.Column(db.String(100))
    books = db.relationship('Book', backref='author', lazy='dynamic')

class Book(db.Model):
    id = db.Column(db.Integer, primary_key=True)
    title = db.Column(db.String(200))
    author_id = db.Column(db.Integer, db.ForeignKey('author.id'))

# Создание объектов
author = Author(name='Лев Толстой')
db.session.add(author)
book1 = Book(title='Война и мир', author=author)
book2 = Book(title='Анна Каренина', author=author)
db.session.add_all([book1, book2])
db.session.commit()

# Запрос книг автора
books = author.books.all()
for b in books:
    print(b.title)

Война и мир
Анна Каренина

2. Многие-ко-многим с помощью вспомогательной таблицы

tags_articles = db.Table('tags_articles',
    db.Column('tag_id', db.Integer, db.ForeignKey('tag.id'), primary_key=True),
    db.Column('article_id', db.Integer, db.ForeignKey('article.id'), primary_key=True)
)

class Tag(db.Model):
    id = db.Column(db.Integer, primary_key=True)
    name = db.Column(db.String(50))

class Article(db.Model):
    id = db.Column(db.Integer, primary_key=True)
    title = db.Column(db.String(100))
    tags = db.relationship('Tag', secondary=tags_articles, backref='articles', lazy='subquery')

# Использование
article = Article(title='Flask и SQLAlchemy')
tag1 = Tag(name='Python')
tag2 = Tag(name='ORM')
article.tags = [tag1, tag2]
db.session.add(article)
db.session.commit()

3. Использование hybrid_property для виртуальных полей

from sqlalchemy.ext.hybrid import hybrid_property

class Employee(db.Model):
    id = db.Column(db.Integer, primary_key=True)
    first_name = db.Column(db.String(50))
    last_name = db.Column(db.String(50))
    salary = db.Column(db.Integer)

    @hybrid_property
    def full_name(self):
        return f'{self.first_name} {self.last_name}'

    @hybrid_property
    def annual_salary(self):
        return self.salary * 12

emp = Employee(first_name='Иван', last_name='Петров', salary=50000)
db.session.add(emp)
db.session.commit()
print(emp.full_name, emp.annual_salary)

Иван Петров 600000

4. Массовая вставка с использованием bulk_insert

data = [{'name': 'Товар1', 'price': 100},
        {'name': 'Товар2', 'price': 200}]
db.session.execute(db.insert(Product).values(data))
db.session.commit()
# Для SQLAlchemy Core:
from sqlalchemy import insert
stmt = insert(Product).values(data)
with engine.connect() as conn:
    conn.execute(stmt)
    conn.commit()

Массовая вставка быстрее, чем добавление по одному объекту, но не вызывает событий ORM.

5. Выполнение raw SQL внутри сессии

result = db.session.execute('SELECT * FROM user WHERE id=:uid', {'uid': 1})
row = result.fetchone()
print(row.username)

# Или с использованием текстового выражения:
from sqlalchemy import text
result = db.session.execute(text('SELECT * FROM user WHERE id=:uid'), {'uid': 1})

john

6. Каскадное удаление

class Parent(db.Model):
    id = db.Column(db.Integer, primary_key=True)
    name = db.Column(db.String(50))
    children = db.relationship('Child', cascade='all, delete-orphan', backref='parent')

class Child(db.Model):
    id = db.Column(db.Integer, primary_key=True)
    name = db.Column(db.String(50))
    parent_id = db.Column(db.Integer, db.ForeignKey('parent.id'))

parent = Parent(name='Родитель')
child = Child(name='Ребенок', parent=parent)
db.session.add(parent)
db.session.commit()
db.session.delete(parent)  # Каскадно удалит и child
db.session.commit()

7. Использование контекстного менеджера для сессий (чистый SQLAlchemy)

from contextlib import contextmanager

@contextmanager
def session_scope():
    session = Session()
    try:
        yield session
        session.commit()
    except Exception:
        session.rollback()
        raise
    finally:
        session.close()

with session_scope() as s:
    user = s.query(User).get(1)
    user.name = 'new_name'
# Автоматический commit/rollback

8. Использование декоратора для упрощения работы с сессией в Flask (без расширения)

from functools import wraps
from flask import g, request, current_app

def with_session(f):
    @wraps(f)
    def decorated_function(*args, **kwargs):
        if not hasattr(g, 'session'):
            g.session = Session()
        try:
            result = f(*args, **kwargs)
            g.session.commit()
            return result
        except Exception as e:
            g.session.rollback()
            raise e
        finally:
            g.session.close()
    return decorated_function

@app.route('/session_example')
@with_session
def session_example():
    user = g.session.query(User).first()
    return f'Hello {user.username}'