Flask Jinja2 en SLQAlchemy many-to-many relatie met de voorwaarden...

Disclaimer: Dit is mijn eerste SQLAlchemy project, dus ik ben niet ervaren en kan in deze post verkeerde uitspraken doen.

Tijdens het werken aan mijn eerste Flask / SQLAchemy project (ik gebruik het niet Flask-SQLAlchemy) raakte ik een zeer hoge en dikke muur. Om dit uit te leggen, denk eraan dat je een blog hebt met berichten en tags. Er is een many-to-many relatie tussen berichten en tags: een bericht kan veel tags hebben en een tag kan veel berichten hebben.

De verenigingstafel en modellen:

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) 

Rustig, aardig, geen problemen. In de weergavefunctie geven we berichten door aan het sjabloon:

def posts_list():

    blog_posts = db.query(BlogPost)

    return render_template(
        ...
        blog_posts=blog_posts)

In het Ninja sjabloon staan we te itereren over de berichten en 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 %}

Mooi, elegant! Maar realistisch?

Het probleem: voorwaarden

Onder de meest gestelde SQLAlchemy vragen over bijvoorbeeld stackoverflow vindt u de many-to-many relatie met de omstandigheden. Waarom? Omdat beginnen met SQLAlchemy is verwarrend.

U loopt tegen problemen aan wanneer u een of meer voorwaarden wilt toevoegen. Hoe doe je dit? De gedefinieerde relatie kan niet meer gebruikt worden of kan deze niet meer gebruikt worden? Ik geloofde dat er een eenvoudige oplossing zou zijn om bijvoorbeeld een verwijderde voorwaarde of een filter toe te voegen aan de tagnaam. Bij het zoeken naar oplossingen vond ik deze tekst, zie de eerste link hieronder: Als je een JOIN voorwaarde nodig hebt die afwijkt van wat de relatie() definieert, dan kan je die relatie niet gebruiken om de aansluiting te doen. Je moet het explitly spellen'.

Nee, nee, nee, nee, dit was niet wat ik zocht, het moet mogelijk zijn om dit gemakkelijker te doen. Toen ik meer en meer las werd ik steeds depressiever en moest ik uiteindelijk concluderen dat het niet mogelijk was om dit makkelijker te doen. Het ergste zelfs, om beide te krijgen BlogPost en BlogTag van een query, moet u beide specificeren:

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

Dit zal resulteren in tuples teruggave, niet in een enkel object.

Samenvattend, als je een relatie hebt en de meeste van de tijd de voorwaarden zijn anders dan is er geen nut voor back_populates. Dus vergeet het herhalen zoals hierboven en begin na te denken tuples en andere manieren te bedenken om te bereiken wat je wilt. Ik kwam met twee mogelijke oplossingen:

- Enkele query met buitenste
voegen - Twee query's - SELECTalleen, eerste query het selecteren van de berichten, tweede query het selecteren van tags voor elk bericht.

Hieronder onderzoek ik beide.

Oplossing 1: Enkele query met buitenste voegen

De query levert een lijst op met tuples paren: post-tag. Een voorbeeld van deze vraag is:

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

We gebruiken outerjoin omdat we willen dat alle BlogPost gegevens in het resultaat worden opgenomen. Wanneer een bericht nul tags heeft, is de tag None, wanneer een bericht twee of meer tags heeft, tuples worden er twee of meer teruggestuurd:

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

Dit betekent dat we niet eenvoudigweg kunnen itereren over de tuples in het sjabloon, omdat een post in meer dan één tuplepost kan staan. We moeten wat nabewerking doen voordat we de gegevens aan het sjabloon doorgeven. Een manier om dit te doen is om meerdere identieke blog_posts te itereren tuples en te reduceren tot één blog_post en alle tags voor een blog_post in een lijst te plaatsen:

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 de weergavefunctie geven we de blog_posts en blog_tags door aan de template:

def posts_list():

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

In het sjabloon doen we iets dergelijks:

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

Oplossing 2: twee query's - SELECTalleen de eerste query's - de eerste query's selecteren de berichten en de tweede query's selecteren de tags voor elk bericht.

Waarom twee vragen in plaats van één vraag? Is dit niet altijd langzamer? Hoewel ik van mening ben dat je altijd moet proberen het aantal vragen tot een minimum te beperken, is het prestatieverschil moeilijk te voorspellen. Sommige vragen zijn veel sneller dan andere. Query caching kan ook de prestaties drastisch verhogen. Voor een website, denk nooit aan de prestaties van een enkele bezoeker, denk altijd na over wat er gebeurt als veel bezoekers verbinding maken, op een link klikken, enz.

De eerste - SELECTalleen de vraag krijgt alle berichten:

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

De tweede SELECT-alleen een query bestaat uit een selecte groep met alle voorwaarden op zijn plaats:

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

Deze query komt alleen terug die aan alle criteria tuples voldoet, wat betekent dat een bericht zonder tags niet in het resultaat zal verschijnen en dat een tag nooit kan worden None. Net als in oplossing 1 doen we de post-processing door een woordenboek aan te maken met een sleutelblog_post_id en een waarde een lijst met de tags met standaarddict. Dit is nu veel eenvoudiger dan oplossing 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 de weergavefunctie geven we de gegevens door zoals we in oplossing 1 hebben gedaan:

def posts_list():

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

Het sjabloon wijkt niet af van oplossing 1. De sjabloon verschilt niet van oplossing 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 %}

Samenvatting

Vergeet de mooie, elegante oplossing die in het begin werd genoemd, deze bestaat alleen in zeer specifieke situaties. Echte toepassingen vereisen vaak vragen met meerdere dynamische condities.

Met Flask, Jinja2 en SQLAlchemy denk in tuples want dit zijn de eenheden die vaak terug zullen komen. Deze hebben enige nabewerking tuples nodig om objecten te maken die geschikt zijn om in het sjabloon te worden weergegeven.

Ik hou niet van deze post-processing, naar mijn mening moet een query terug te keren alle gegevens die kunnen worden gezet in een sjabloon, maar op dit moment weet ik niet op een andere manier. Ik heb geen voorkeur voor een van de twee bovenstaande oplossingen, hoewel ik de neiging heb om voor de twee vragen te kiezen - SELECTalleen voor de eenvoud ervan. Voorlopig zal ik dit gebruiken.

Voor het geval u dit wilt proberen, hier is de volledige 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 / credits

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