Описание связи many-to-many (многие ко многим) в SQLAlchemy. Пример создания таблиц и сохранения данных.
Более новая статья по теме связей relationship SQLAlchemy, более точная, но в чем-то не заменяющая текущую.
Описание моделей (классов)
Мы создаем два класса: Task и Category, в одной таблице будут сохранятся задачи, в другой категории.
Таблица задач
Таблица категорий
У одной задачи может быть несколько категорий. Для этих целей наилучшим образом подойдет сохранение связи между задачей и категорией в отдельной таблице. Это будет третья таблица, которая будет создана с помощью связи db.relationship.
Все это делается описанием классов моделей таким образом:
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 надо добавить пару дополнительных свойств такого плана:
'Category',
secondary=tasks_categories,
backref=db.backref('task', lazy='dynamic'),
single_parent=True,
cascade="all, delete, delete-orphan"
)
Однако в данном примере это не подходит, иначе удалив задачу с привязанной к ней категорией, мы удалим эти категории из всех других задач в том числе.
Как сохранить данные
Пример ниже представляет из себя методы с роутом (через блюпринт). Она получает данные через метод POST. Затем эти данные сохраняются в БД и возвращаются обратно.
Внутри методов осуществляется сохранение задачи, категорий и связи с ними. Все основные моменты пояснены в первом методе.
Метод сохранения к задаче одной категории
@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()
Метод сохранения к задаче несколько категорий
В этом случае наш метод немного изменится. С фронта мы получаем список категорий через запятую. Формируем из этой строки список категорий. Проверяем каждую категорию на существование в базе данных. Затем или добавляем к задаче её или создаём и опять же добавляем к задаче.
@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()
Как получить задачи по категориям
Мы должны в класс категорий:
__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))
добавить связь:
'Task',
secondary=categories_task,
backref=db.backref('tasks', lazy='dynamic')
)
Теперь мы можем получать задачи по id категории так:
Ответ:
Получение сериализованных данных
Также эти данные можно получить сериализованными (обработанными).
Запросим обработанные данные:
Для этого в метод Task (не перепутайте) надо добавить обработку:
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]
}
Ответ будет таким:
'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'}]
Еще один пример выборки many to many
'categories_task',
db.Column('task_id', db.Integer(), db.ForeignKey('task.id')),
db.Column('cat_id', db.Integer(), db.ForeignKey('categories.id'))
)
Вот как можно отобрать task-и из категорий:
# Получим все task присоединенные к категориям
response = db.session.query(categories_task).filter_by(cat_id=cat_id).all()
# Затем по категориям отберем task
tasks = []
for x in response:
task = db.session.query(Task).filter_by(id=x.task_id).first().serialize
tasks.append(task)
return json.dumps(tasks)
Из участка выше, это:
for x in response:
task = db.session.query(Task).filter_by(id=x.task_id).first().serialize
tasks.append(task)
Можно также записать так:
query = db.session.query(Task).filter_by(user_id=user_id).order_by(db.desc(Task.time))
# Заберем все tasks с учетом параметров пагинации
tasks = [x.serialize for x in query.offset(offset).limit(limit)]
Только изучаю python вообще и flask в частности, возник вопрос по коду:
># Вернем обновленные данные обратно на фронт
> return get_items()
Нет перенаправления и рендера, значит ответ функции обрабатывается через ajax скриптами на фронте? Изначально подумал, что просто вызов из другого места, но тогда не было бы роута в заголовке функции.
Или это придуманный пример и полная функциональность не прописана изначально?
Код рабочий, сокращенный вариант, вырезанный из рабочей структуры.
return get_items() — в данном примере это не важно и поэтому не расписано. Эта ссылка на функцию получения всех элементов в БД, чтобы отправить на фронт. Можно ничего не отправлять или отправить что-то другое, зависит от того, что ожидает фронт.
В этом примере приведен только бэк, потому что фронт может быть на любой технологии и это никак не изменит бэк. У меня, например, со стороны фронта этот код обрабатывается на svelte.
По flask написан большой гайд. Может найдете что-то полезное.