SQLAlchemy — many-to-many

Описание связи many-to-many (многие ко многим) в SQLAlchemy. Пример создания таблиц и сохранения данных.

Описание моделей (классов)

Мы создаем два класса: Task и Category, в одной таблице будут сохранятся задачи, в другой категории.

Таблица задач

Таблица категорий

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

Все это делается описанием классов моделей таким образом:

# ploshadka.net
from datetime import datetime
from flask_login import UserMixin
from flask_security import RoleMixin

tasks_categories = db.Table(
'tasks_categories',
db.Column('task_id', db.Integer(), db.ForeignKey('task.id')),
db.Column('cat_id', db.Integer(), db.ForeignKey('categories.id'))
)

class Task(db.Model, RoleMixin):
__tablename__ = 'tasks'
id = db.Column(db.Integer(), primary_key=True, unique=True)
user_id = db.Column(db.Integer, db.ForeignKey('users.id'))
name = db.Column(db.String(255))
time = db.Column(db.DateTime())

# Для получения доступа к связанным объектам
cats = db.relationship('Category', secondary=tasks_categories, backref=db.backref('tasks', lazy='dynamic'))

class Category(db.Model, RoleMixin):
__tablename__ = 'categories'
id = db.Column(db.Integer(), primary_key=True, unique=True)
user_id = db.Column(db.Integer, db.ForeignKey('users.id'))
name = db.Column(db.String(255))

Если требуется в дальнейшем удалять одной командой все связанные данные, то в db.relationship надо добавить пару дополнительных свойств такого плана:

cats = db.relationship(
'Category',
secondary=tasks_categories,
backref=db.backref('task', lazy='dynamic'),
single_parent=True,
cascade="all, delete, delete-orphan"
)

Однако в данном примере это не подходит, иначе удалив задачу с привязанной к ней категорией, мы удалим эти категории из всех других задач в том числе.

Как сохранить данные

Пример ниже представляет из себя методы с роутом (через блюпринт). Она получает данные через метод POST. Затем эти данные сохраняются в БД и возвращаются обратно.

Внутри методов осуществляется сохранение задачи, категорий и связи с ними. Все основные моменты пояснены в первом методе.

Метод сохранения к задаче одной категории

# ploshadka.net
@task.route('/add-new-item/', methods=['POST'])
@login_required
def add_new_item():

# Данные с фронта
data = request.get_json()
name = data['name']
category = data['category']

# Текущий пользователь
user_id = flask_login.current_user.id

# Определим существование категории
c1 = db.session.query(Category).filter_by(user_id=user_id, name=category).first()

# Если категории нет, создадим её
if c1 is None:
c1 = Category(user_id=user_id, name=category)
db.session.add(c1)

# Используем flush, чтобы получить id категории, которая будет добавлена
db.session.flush()

# Добавим задачу
new_task = Task(user_id=user_id, name=name, time=datetime.now())

# Добавим связь
new_task.cats.append(c1)
db.session.add(new_task)

# Теперь сохраним все что выше в нашу базу данных
db.session.commit()

# Вернем обновленные данные обратно на фронт
return get_items()

Метод сохранения к задаче несколько категорий

В этом случае наш метод немного изменится. С фронта мы получаем список категорий через запятую. Формируем из этой строки список категорий. Проверяем каждую категорию на существование в базе данных. Затем или добавляем к задаче её или создаём и опять же добавляем к задаче.

# ploshadka.net
@task_bp.route('/add-new-item/', methods=['POST'])
@login_required
def add_new_item():
user_id = flask_login.current_user.id

# Получим данные с фронта
data = request.get_json()
name = data['name']
categories = data['categories']

# Добавим новую задачу
new_task = Task(user_id=user_id, name=name, time=datetime.now())

# Сделаем лист из нескольких категорий (разделим их по запятым)
categories = [x.strip() for x in categories.split(',')]

# Для каждой категории
for category_name in categories:

# Определим существует ли категория
category_in_db = db.session.query(Category).filter_by(user_id=user_id, name=category_name).first()

if category_in_db:
new_task.cats.append(category_in_db)
else:
c = Category(user_id=user_id, name=category_name)
db.session.add(c)
db.session.flush()
new_task.cats.append(c)

# Сохраним
db.session.add(new_task)
db.session.commit()

# Вернем обновленные данные
return get_items()

Как получить задачи по категориям

Мы должны в класс категорий:

class Category(db.Model, RoleMixin):
__tablename__ = 'categories'
id = db.Column(db.Integer(), primary_key=True, unique=True)
user_id = db.Column(db.Integer, db.ForeignKey('users.id'))
name = db.Column(db.String(255))

добавить связь:

tasks = db.relationship(
'Task',
secondary=categories_done,
backref=db.backref('tasks', lazy='dynamic')
)

Теперь мы можем получать задачи по id категории так:

response = db.session.query(Category).filter_by(id=45).first().tasks

Ответ:

[<Done 69>, <Done 70>]

Получение сериализованных данных

Также эти данные можно получить сериализованными (обработанными).

Больше о серилизации данных можно узнать в статьях получение списков из SQLAlchemy и в примерах запросов на SQLAlchemy.

Запросим обработанные данные:

response = [x.serialize for x in db.session.query(Category).filter_by(id=45).first().tasks]

Для этого в метод Task (не перепутайте) надо добавить обработку:

@property
def serialize(self):
return {
'id': self.id,
'time': self.time.strftime('%H:%M'),
'time_min': self.time.strftime('%M'),
'time_hour': self.time.strftime('%H'),
'date': self.time.strftime('%Y-%m-%d'),
'name': self.name,
'cats': [x.get_name for x in self.cats]
}

Ответ будет таким:

[{'cats': [{'id': 45, 'name': 'солнце'}],
'date': '2021-02-19',
'id': 69,
'name': 'тест 1',
'time': '12:27',
'time_hour': '12',
'time_min': '27'},
{'cats': [{'id': 45, 'name': 'солнце'}],
'date': '2021-02-19',
'id': 70,
'name': 'тест 2',
'time': '12:28',
'time_hour': '12',
'time_min': '28'}]