SQLAlchemy — many-to-many

Admin SQLAlchemy

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

Более новая статья по теме связей relationship 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_task,
    backref=db.backref('tasks', lazy='dynamic')
)

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

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

Ответ:

[<task 69>, <task 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'}]

Еще один пример выборки many to many

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

Вот как можно отобрать task-и из категорий:

def get_tasks_by_category(cat_id):
   
 # Получим все 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)

Из участка выше, это:

tasks = []
    for x in response:
        task = db.session.query(Task).filter_by(id=x.task_id).first().serialize
        tasks.append(task)

Можно также записать так:

# Сформируем основной запрос по tasks
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)]

У сайта нет цели самоокупаться, поэтому на сайте нет рекламы. Но если вам пригодилась информация, можете лайкнуть страницу, оставить комментарий или отправить мне подарок на чашечку кофе.

Комментарии к статье “SQLAlchemy — many-to-many” (2)

  • Алексей
    13.04.2021 в 18:56

    Только изучаю python вообще и flask в частности, возник вопрос по коду:

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

    Нет перенаправления и рендера, значит ответ функции обрабатывается через ajax скриптами на фронте? Изначально подумал, что просто вызов из другого места, но тогда не было бы роута в заголовке функции.

    Или это придуманный пример и полная функциональность не прописана изначально?

    • Admin
      Автор записи
      15.04.2021 в 12:16

      Код рабочий, сокращенный вариант, вырезанный из рабочей структуры.

      return get_items() — в данном примере это не важно и поэтому не расписано. Эта ссылка на функцию получения всех элементов в БД, чтобы отправить на фронт. Можно ничего не отправлять или отправить что-то другое, зависит от того, что ожидает фронт.

      В этом примере приведен только бэк, потому что фронт может быть на любой технологии и это никак не изменит бэк. У меня, например, со стороны фронта этот код обрабатывается на svelte.

      По flask написан большой гайд. Может найдете что-то полезное.

Добавить комментарий

Напишите свой комментарий, если вам есть что добавить/поправить/спросить по теме текущей статьи:
"SQLAlchemy — many-to-many"