Flask Jinja2 und SLQAlchemy many-to-many Beziehung mit Bedingungen

Haftungsausschluss: Dies ist mein erstes SQLAlchemy Projekt, daher bin ich nicht erfahren und kann in diesem Beitrag falsche Aussagen machen.

Während der Arbeit an meinem ersten Flask / SQLAchemy-Projekt (ich benutze Flask-SQLAlchemyes nicht) traf ich auf eine sehr hohe und dicke Wand. Um dies zu erklären, solltest du einen Blog mit Beiträgen und Tags haben. Es gibt eine many-to-many Beziehung zwischen Beiträgen und Tags: Ein Beitrag kann viele Tags haben und ein Tag kann viele Beiträge haben.

Die Zuordnungstabelle und die Modelle:

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) 

Einfach, schön, keine Probleme. In der View-Funktion übergeben wir Beiträge an die Vorlage:

def posts_list():

    blog_posts = db.query(BlogPost)

    return render_template(
        ...
        blog_posts=blog_posts)

In der Ninja-Vorlage iterieren wir über die Beiträge und 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.blog_tags %}
                {{ blog_post.blog_tags|join(', ', attribute='name') }}
            {% endif %}
            </td>
        </tr>
    {% endfor %}

Wunderschön, elegant! Aber realistisch?

Das Problem: Bedingungen

Zu den am häufigsten gestellten SQLAlchemy Fragen, z.B. zu Stackoverflow, gehört der many-to-many Zusammenhang mit Bedingungen. Warum? Denn der Anfang mit SQLAlchemy ist verwirrend.

Sie stoßen auf Probleme, wenn Sie eine oder mehrere Bedingungen hinzufügen möchten. Wie machst du das? Die definierte Beziehung kann nicht mehr verwendet werden oder nicht? Ich dachte, es gäbe eine einfache Lösung, um z.B. eine gelöschte Bedingung oder einen Filter auf den Tag-Namen hinzuzufügen. Auf der Suche nach Lösungen habe ich diesen Text gefunden, siehe den ersten Link unten: Wenn Sie eine JOIN-Bedingung benötigen, die sich von dem unterscheidet, was die relationship() definiert, dann können Sie diese Beziehung nicht verwenden, um den Join durchzuführen. Du musst es explizit buchstabieren".

Nein, nein, nein, nein, nein, das war nicht das, wonach ich gesucht habe, es muss möglich sein, das einfacher zu machen. Als ich immer mehr las, wurde ich immer deprimierter und musste schließlich feststellen, dass es nicht möglich war, dies einfacher zu machen. Am schlimmsten ist es sogar, dass Sie beide angeben müssen, um beides zu erhalten BlogPost und BlogTag von einer Abfrage zu profitieren:

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

Dies führt dazu tuples , dass kein einziges Objekt zurückgegeben wird.

Zusammenfassend lässt sich sagen, wenn Sie eine Beziehung haben und die meisten der Zeit die Bedingungen anders sind, dann gibt es keinen Sinn für back_populates. Also vergiss besser die Iteration wie oben und fang an zu denken tuples und andere Wege zu finden, um das zu erreichen, was du willst:

- Einzelne Abfrage mit Outerjoins
- Zwei Abfragen - SELECTnur die erste Abfrage wählt die Beiträge aus, die zweite Abfrage wählt die Tags für jeden Beitrag aus.

Unten untersuche ich beides.

Lösung 1: Einzelabfrage mit Outerjoins

Die Abfrage gibt eine Liste von tuples Paaren zurück: post-tag. Ein Beispiel für diese Abfrage ist:

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))

Wir verwenden outerjoin, weil wir wollen, dass alle BlogPost Einträge im Ergebnis erscheinen. Wenn ein Beitrag keine Tags hat, ist Nonedas Tag, wenn ein Beitrag zwei oder mehr Tags hat, werden zwei oder mehr tuples zurückgegeben:

[(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'))]

Das bedeutet, dass wir nicht einfach über das tuples in der Vorlage iterieren können, da ein Beitrag in mehr als einem tuplesein kann. Wir müssen eine Nachbearbeitung durchführen, bevor wir die Daten an die Vorlage übergeben. Eine Möglichkeit, dies zu tun, ist, über die tuples und reduzieren mehrere identische blog_posts auf einen einzigen blog_post und setzen Sie alle Tags für einen blog_post in eine Liste:

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)

In der View-Funktion übergeben wir die blog_posts und blog_tags an die Vorlage:

def posts_list():

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

In der Vorlage machen wir so etwas wie:

    {% 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 %}

Lösung 2: Zwei Abfragen - SELECTnur die erste Abfrage, die die Beiträge auswählt, und die zweite Abfrage, die die Tags für jeden Beitrag auswählt.

Warum zwei Abfragen statt einer? Ist das nicht immer langsamer? Obwohl ich glaube, dass Sie immer versuchen sollten, die Anzahl der Anfragen zu minimieren, ist der Leistungsunterschied schwer vorherzusagen. Einige Abfragen sind viel schneller als andere. Auch das Zwischenspeichern von Abfragen kann die Performance drastisch steigern. Für eine Website, denken Sie nie an die Leistung eines einzelnen Besuchers, betrachten Sie immer, was passiert, wenn sich viele Besucher verbinden, klicken Sie auf einen Link, etc.

Die erste SELECT-nur Abfrage erhält alle Beiträge:

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

Die zweite SELECT-nur Abfrage besteht aus einem select mit allen vorhandenen Bedingungen:

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))

Diese Abfrage liefert nur Ergebnisse, die alle Kriterien tuples erfüllen, was bedeutet, dass ein Beitrag ohne Tags nicht im Ergebnis angezeigt wird und ein Tag nie sein Nonekann. Wie in Lösung 1 führen wir Nachbearbeitung durch, indem wir ein Dictionary mit der Schlüssel blog_post_id und einem Wert erstellen, der eine Liste mit den Tags mit defaultdict enthält. Dies ist jetzt viel einfacher als Lösung 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)

In der View-Funktion übergeben wir die Daten wie in Lösung 1:

def posts_list():

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

Die Vorlage unterscheidet sich nicht von Lösung 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 %}

Zusammenfassung

Vergessen Sie die eingangs erwähnte schöne, elegante Lösung, sie existiert nur in ganz bestimmten Situationen. Real Life-Anwendungen erfordern oft Abfragen mit mehreren dynamischen Bedingungen.

Mit Flask, Jinja2 und SQLAlchemy denken Sie daran, tuples denn dies sind die Einheiten, die oft zurückgegeben werden. Diese tuples erfordern eine gewisse Nachbearbeitung, um Objekte zu erstellen, die für die Anzeige in der Vorlage geeignet sind.

Ich mag diese Nachbearbeitung nicht, meiner Meinung nach sollte eine Anfrage alle Daten zurückgeben, die in eine Vorlage eingegeben werden können, aber im Moment kenne ich keinen anderen Weg. Ich habe keine Präferenz für eine der beiden obigen Lösungen, obwohl ich dazu tendiere, die beiden Abfragen zu wählen - SELECTnur die Lösung wegen ihrer Einfachheit. Im Moment werde ich das hier verwenden.

Falls Sie dies ausprobieren möchten, hier ist der vollständige Code:

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))

Links / Impressum

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