angle-uparrow-clockwisearrow-counterclockwisearrow-down-uparrow-leftatcalendarcard-listchatcheckenvelopefolderhouseinfo-circlepencilpeoplepersonperson-fillperson-plusphoneplusquestion-circlesearchtagtrashx

Flask Jinja2 и связь SLQAlchemy many-to-many с условиями.

Какой смысл определять отношения с ? back_populatesНачни думать.

17 июня 2019
post main image

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

Во время работы над моим первым Flask / SQLAchemy проектом (я не использую ) я наткнулся на очень высокую и толстую стену. Чтобы объяснить это, подумайте, что у вас есть блог с записями и тегами. Существует many-to-many взаимосвязь между сообщениями и тегами: сообщение может иметь много тегов, а тег может иметь много сообщений.

Ассоциированный стол и модели:

Base = declarative_base()

# many-to-many association table: blog_post - blog_tag
blog_post_mtm_blog_tag_table = Table('blog_post_mtm_blog_tag', Base.metadata,
    Column('blog_post_id', Integer, ForeignKey('blog_post.id')),
    Column('blog_tag_id', Integer, ForeignKey('blog_tag.id'))
)

class BlogPost(Base):
    __tablename__ = 'blog_post'

    id = Column(Integer, primary_key=True)
    deleted = Column(Boolean, default=False)
    title = Column(String)
    # many-to-many relationship with child
    blog_tags = relationship(
        'BlogTag', 
        secondary=blog_post_mtm_blog_tag_table,
        back_populates='blog_posts')

    def __repr__(self):
        return "%s(title=%r)" % (self.__class__.__name__, self.title) 


class BlogTag(Base):
    __tablename__ = 'blog_tag'

    id = Column(Integer, primary_key=True)
    deleted = Column(Boolean, default=False)
    name = Column(String)
    # many-to-many relationship with parent
    blog_posts = relationship('BlogPost',
        secondary=blog_post_mtm_blog_tag_table,
        back_populates='blog_tags')

    def __repr__(self):
        return "%s(name=%r)" % (self.__class__.__name__, self.name) 

Полегче, полегче, спокойнее, без проблем. В функции просмотра мы передаем сообщения в шаблон:

def posts_list():

    blog_posts = db.query(BlogPost)

    return render_template(
        ...
        blog_posts=blog_posts)

В шаблоне Ninja мы итерируем сообщения и теги:

    {% for blog_post in blog_posts %}
        <tr>
            <td>
                <a href="{{ url_for('blog.posts_post_edit', id=blog_post.id) }}">
                    <i class="fa fa-pencil-alt"></i> {{ blog_post.title }} 
                </a>
            </td>
            <td>
            {% if blog_post.blog_tags %}
                {{ blog_post.blog_tags|join(', ', attribute='name') }}
            {% endif %}
            </td>
        </tr>
    {% endfor %}

Красиво, элегантно! Но реалистично?

Проблема: условия

Среди SQLAlchemy вопросов, наиболее часто рассматриваемых, например, при переполнении стека, вы найдете many-to-many связь с условиями. Почему? Потому что начинать с этого неудобно.

Вы сталкиваетесь с проблемами, когда хотите добавить одно или несколько условий. Как ты это делаешь? Определенные отношения больше не могут быть использованы или могут быть использованы? Я верил, что будет простое решение, чтобы добавить, например, удаленное условие или фильтр по имени тега. В поисках решений я нашел этот текст, см. первую ссылку ниже: Если вам нужно условие JOIN, отличное от того, что определяет функция Relations(), то вы не можете использовать это условие для присоединения. Ты должен объяснить это по буквам".

Нет, нет, нет, нет, нет, это не то, что я искал, это должно быть возможно сделать проще. Читая все больше и больше, я впадал в депрессию и, наконец, пришел к выводу, что сделать это легче не представляется возможным. Хуже того, чтобы получить и то, и другое, BlogPost и другое из запроса, нужно указать и то, и другое:

qry = db.query(BlogPost, BlogTag)......

Это приведет к тому, что будет возвращен ни один предмет.

Подводя итог, если у вас есть отношения и большая часть времени условия отличаются, то нет смысла в том, чтобы . back_populates.. Поэтому лучше забудьте об итерации, как описано выше, и начните думать и искать другие способы достижения желаемого: я придумал два возможных решения:

- Одиночный запрос с внешними соединениями - Два запроса - SELECTтолько первый запрос, выбор сообщений, второй запрос, выбор тегов для каждого сообщения.

Ниже я расследую и то, и другое.

Решение 1: Одиночный запрос с внешними соединениями

Запрос возвращает список, tuples состоящий из пар: posttag. Примером такого запроса может служить:

qry = db.query(BlogPost, BlogTag) \
    .outerjoin(blog_post_mtm_blog_tag_table) \
    .outerjoin(BlogTag, and_( (BlogTag.id == blog_post_mtm_blog_tag_table.c.blog_tag_id) )) \
    .order_by(asc(BlogPost.title)) \
    .order_by(asc(BlogTag.name))

Мы используем внешний шарнир, потому что хотим, чтобы в результате появились все BlogPost записи. Если сообщение имеет нулевые теги, то тегом является None, если сообщение имеет два или более тегов, возвращается два или более:

[(BlogPost(title='Drinks on the house'), BlogTag(name='refreshment')), (BlogPost(title='Hello Mexico'), BlogTag(name='Holiday')), (BlogPost(title='Hello Mexico'), BlogTag(name='Mexico')), (BlogPost(title='Mountain shoes'), None), (BlogPost(title='Sunny beaches'), BlogTag(name='Holiday')), (BlogPost(title='Sunny beaches'), BlogTag(name='Travel'))]

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

blog_posts = []
blog_post_id2blog_tags = {}
for blog_post, blog_tag in qry.all():
    # add blog_post
    if not blog_post in blog_posts:
        blog_posts.append(blog_post)

    if blog_tag == None:
        if not blog_post.id in blog_post_id2blog_tags:
            blog_post_id2blog_tags[blog_post.id] = []
        continue
    if not blog_post.id in blog_post_id2blog_tags:
        blog_post_id2blog_tags[blog_post.id] = []
    blog_post_id2blog_tags[blog_post.id].append(blog_tag)

В функции просмотра мы передаем в шаблон посты и блоги blog_posts и blog_tags:

def posts_list():

    ... 
    
    return render_template(
        ...
        blog_posts=blog_posts,
        blog_post_id2blog_tags=blog_post_id2blog_tags)

В шаблоне мы делаем что-то вроде:

    {% for blog_post in blog_posts %}
        <tr>
            <td>
                <a href="{{ url_for('blog.posts_post_edit', id=blog_post.id) }}">
                    <i class="fa fa-pencil-alt"></i> {{ blog_post.title }} 
                </a>
            </td>
            <td>
            {% if blog_post.id in blog_post_id2blog_tags %}
                {{ blog_post2blog_tags[blog_post.id]|join(', ', attribute='name') }}
            {% endif %}
            </td>
        </tr>
    {% endfor %}

Решение 2: Два запроса - SELECTтолько первый запрос, выбирающий сообщения, и второй запрос, выбирающий теги для каждого сообщения.

Почему два запроса вместо одного? Разве это не всегда медленнее? Хотя я считаю, что вы всегда должны стараться минимизировать количество запросов, разницу в производительности трудно предсказать. Некоторые запросы выполняются намного быстрее, чем другие. Кэширование запросов также может значительно увеличить производительность. Для веб-сайта, никогда не думайте о производительности одного посетителя, всегда думайте о том, что происходит, когда много посетителей соединяются, щелкните ссылку и т.д.

Первый - SELECTтолько запрос получает все сообщения:

blog_posts = db.query(BlogPost) \
    .order_by(asc(BlogPost.title))

Второй - SELECTтолько запрос состоит из выбора со всеми существующими условиями:

qry = db.query(BlogPost.id, BlogTag) \
    .filter( and_( \
        (BlogPost.id == blog_post_mtm_blog_tag_table.c.blog_post_id), \
        (BlogTag.id == blog_post_mtm_blog_tag_table.c.blog_tag_id) \
        )) \
    .order_by(asc(BlogPost.title)) \
    .order_by(asc(BlogTag.name))

Этот запрос возвращает только сообщения, tuples отвечающие всем критериям, что означает, что сообщение без тегов не будет отображаться в результате, а тег никогда не может быть . Как и в решении 1, мы выполняем постобработку, создавая словарь с ключом blog_post_id и значением списка, содержащего теги с использованием стандартного знака. Теперь это намного проще, чем решение 1:

from collections import defaultdict

blog_post_id2blog_tags = defaultdict(list)
for blog_post_id, blog_tag in qry.all():
    blog_post_id2blog_tags[blog_post_id].append(blog_tag)

В функции просмотра мы передаем данные, как мы это сделали в решении 1:

def posts_list():

    ... 
    
    return render_template(
        ...
        blog_posts=blog_posts,
        blog_post_id2blog_tags=blog_post_id2blog_tags)

Шаблон не отличается от решения 1:

    {% for blog_post in blog_posts %}
        <tr>
            <td>
                <a href="{{ url_for('blog.posts_post_edit', id=blog_post.id) }}">
                    <i class="fa fa-pencil-alt"></i> {{ blog_post.title }} 
                </a>
            </td>
            <td>
            {% if blog_post.id in blog_post_id2blog_tags %}
                {{ blog_post2blog_tags[blog_post.id]|join(', ', attribute='name') }}
            {% endif %}
            </td>
        </tr>
    {% endfor %}

Резюме

Забудьте о красивом, элегантном решении, упомянутом в начале, оно существует только в очень специфических ситуациях. Реальные приложения часто требуют запросов с несколькими динамическими условиями.

С, Jinja2 и SQLAlchemy думать в, tuples потому что это те единицы, которые часто будут возвращены. Они tuples требуют некоторой постобработки для создания объектов, подходящих для отображения в шаблоне.

Мне не нравится эта постобработка, на мой взгляд, запрос должен возвращать все данные, которые можно поместить в шаблон, но на данный момент я не знаю другого пути. Я не отдаю предпочтения ни одному из двух вышеперечисленных решений, хотя, как правило, выбираю два запроса - SELECTтолько для его простоты. На данный момент я воспользуюсь этим.

На случай, если вы захотите попробовать, вот полный код:

from sqlalchemy import Table, Column, Integer, String, Boolean, BigInteger, DateTime, ForeignKey, func, and_, or_, desc, asc, create_engine, inspect, sql
from sqlalchemy.orm import relationship, Session, with_polymorphic, backref, contains_eager, Query
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.sql import func, label

import os
import sys


Base = declarative_base()

# many-to-many association table: blog_post - blog_tag
blog_post_mtm_blog_tag_table = Table('blog_post_mtm_blog_tag', Base.metadata,
    Column('blog_post_id', Integer, ForeignKey('blog_post.id')),
    Column('blog_tag_id', Integer, ForeignKey('blog_tag.id'))
)

class BlogPost(Base):
    __tablename__ = 'blog_post'

    id = Column(Integer, primary_key=True)
    deleted = Column(Boolean, default=False)
    title = Column(String)
    # many-to-many relationship with child
    blog_tags = relationship(
        'BlogTag', 
        secondary=blog_post_mtm_blog_tag_table,
        back_populates='blog_posts')

    def __repr__(self):
        return "%s(title=%r)" % (self.__class__.__name__, self.title) 


class BlogTag(Base):
    __tablename__ = 'blog_tag'

    id = Column(Integer, primary_key=True)
    deleted = Column(Boolean, default=False)
    name = Column(String)
    # many-to-many relationship with parent
    blog_posts = relationship('BlogPost',
        secondary=blog_post_mtm_blog_tag_table,
        back_populates='blog_tags')

    def __repr__(self):
        return "%s(name=%r)" % (self.__class__.__name__, self.name) 


# not show/show generated sql
engine = create_engine('sqlite://')
#engine = create_engine('sqlite://', echo=True)

Base.metadata.create_all(engine)

from sqlalchemy.orm import sessionmaker
session = sessionmaker()
session.configure(bind=engine)
db = session()

# posts
post_hello_mexico = BlogPost(title='Hello Mexico')
post_sunny_beach = BlogPost(title='Sunny beaches')
post_mountain_shoes = BlogPost(title='Mountain shoes')
post_drinks_on_the_house = BlogPost(title='Drinks on the house')

# tags
tag_mexico = BlogTag(name='Mexico')
tag_holiday = BlogTag(name='Holiday')
tag_travel = BlogTag(name='Travel')
tag_refreshment = BlogTag(name='refreshment')
tag_sandy = BlogTag(name='Sandy')

# post tags
post_hello_mexico.blog_tags.append(tag_mexico)
post_hello_mexico.blog_tags.append(tag_holiday)
post_sunny_beach.blog_tags.append(tag_travel)
post_sunny_beach.blog_tags.append(tag_holiday)
post_drinks_on_the_house.blog_tags.append(tag_refreshment)

db.add_all([post_hello_mexico, post_sunny_beach, post_mountain_shoes, post_drinks_on_the_house, 
    tag_mexico, tag_holiday, tag_travel, tag_refreshment, tag_sandy])
db.commit()


print("Beautiful, elegant, but realistic?")
qry = db.query(BlogPost) \
    .order_by(asc(BlogPost.title))

for blog_post in qry.all():
    print("blog_post {}".format(blog_post.title))
    if blog_post.blog_tags:
        for blog_tag in blog_post.blog_tags:
            print("- blog_tag = {}".format(blog_tag.name))


print("\n\nSolution 1: Single query with outerjoins")

print("Solution 1: query")
qry = db.query(BlogPost, BlogTag) \
    .outerjoin(blog_post_mtm_blog_tag_table) \
    .outerjoin(BlogTag, and_( (BlogTag.id == blog_post_mtm_blog_tag_table.c.blog_tag_id) )) \
    .order_by(asc(BlogPost.title)) \
    .order_by(asc(BlogTag.name))

print("Solution 1: qry.all() = {}".format(qry.all()))

print("Solution 1: create blog_posts and blog_tags dict")
blog_posts = []
blog_post_id2blog_tags = {}
for blog_post, blog_tag in qry.all():
    # add blog_post
    if not blog_post in blog_posts:
        blog_posts.append(blog_post)

    if blog_tag == None:
        if not blog_post.id in blog_post_id2blog_tags:
            blog_post_id2blog_tags[blog_post.id] = []
        continue
    if not blog_post.id in blog_post_id2blog_tags:
        blog_post_id2blog_tags[blog_post.id] = []
    blog_post_id2blog_tags[blog_post.id].append(blog_tag)

print("Solution 1: template")
for blog_post in blog_posts:
    print("blog_post {}".format(blog_post.title))
    if blog_post.id in blog_post_id2blog_tags:
        for blog_tag in blog_post_id2blog_tags[blog_post.id]:
            print("- blog_tag = {}".format(blog_tag.name))


print("\n\nSolution 2: Two query select-only, the first query selecting the posts, and the second query selecting tags for each post")

print("Solution 2: select-query-1 blog_posts")
blog_posts = db.query(BlogPost) \
    .order_by(asc(BlogPost.title))

print("Solution 2: select-query-2 blog_tags for all blog_posts")
qry = db.query(BlogPost.id, BlogTag) \
    .filter( and_( \
        (BlogPost.id == blog_post_mtm_blog_tag_table.c.blog_post_id), \
        (BlogTag.id == blog_post_mtm_blog_tag_table.c.blog_tag_id) \
        )) \
    .order_by(asc(BlogPost.title)) \
    .order_by(asc(BlogTag.name))

print("Solution 2: qry.all() = {}".format(qry.all()))

print("Solution 2: create blog_tags dict")
from collections import defaultdict

blog_post_id2blog_tags = defaultdict(list)
for blog_post_id, blog_tag in qry.all():
    blog_post_id2blog_tags[blog_post_id].append(blog_tag)

print("Solution 2: template")
for blog_post in blog_posts:
    print("blog_post {}".format(blog_post.title))
    if blog_post.id in blog_post_id2blog_tags:
        for blog_tag in blog_post_id2blog_tags[blog_post.id]:
            print("- blog_tag = {}".format(blog_tag.name))

Ссылки / кредиты

Confused about Many-to-Many outerjoin
https://groups.google.com/forum/?fromgroups=#!topic/sqlalchemy/GNIBQMvMRg8

SqlAlchemy - Many to Many outer join with conditions on join
https://stackoverflow.com/questions/22909531/sqlalchemy-many-to-many-outer-join-with-conditions-on-join

SQLAlchemy: complex ON clause when performing LEFT JOIN on many-to-many relation
https://stackoverflow.com/questions/25868681/sqlalchemy-complex-on-clause-when-performing-left-join-on-many-to-many-relation

SQLAlchemy: querying all objects with no direct relationship
https://stackoverflow.com/questions/47918946/sqlalchemy-querying-all-objects-with-no-direct-relationship

Подробнее

SQLAlchemy

Оставить комментарий

Комментируйте анонимно или войдите в систему, чтобы прокомментировать.

Комментарии

Оставьте ответ

Ответьте анонимно или войдите в систему, чтобы ответить.