Flask Jinja2 et SLQAlchemy many-to-many relation avec les conditions

Avertissement : C'est mon premier SQLAlchemy projet donc je ne suis pas expérimenté et je peux faire de fausses déclarations dans ce message.

En travaillant sur mon premier Flask projet / SQLAchemy (je n'utilise pas Flask-SQLAlchemy) j'ai heurté un mur très haut et épais. Pour expliquer cela, considérez que vous avez un blog avec des messages et des tags. Il y a une many-to-many relation entre les messages et les balises : un message peut avoir plusieurs balises et une balise peut avoir plusieurs messages.

La table d'association et les modèles :

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) 

Facile, sympa, pas de problème. Dans la fonction de vue, nous passons les messages au modèle :

def posts_list():

    blog_posts = db.query(BlogPost)

    return render_template(
        ...
        blog_posts=blog_posts)

Dans le gabarit Ninja, nous itérons sur les messages et les balises :

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

Magnifique, élégant ! Mais réaliste ?

Le problème : les conditions

Parmi les SQLAlchemy questions les plus consultées, par exemple sur le débordement de pile, vous trouverez la many-to-many relation avec les conditions. Pourquoi ? Parce que commencer par SQLAlchemy est déroutant.

Vous rencontrez des problèmes lorsque vous voulez ajouter une ou plusieurs conditions. Comment faites-vous cela ? La relation définie ne peut plus être utilisée ou peut-elle l'être ? Je pensais qu'il y aurait une solution simple pour ajouter par exemple une condition supprimée ou un filtre sur le nom de l'étiquette. En cherchant des solutions, j'ai trouvé ce texte, voir le premier lien ci-dessous : Si vous avez besoin d'une condition JOIN différente de ce que définit la relation(), alors vous ne pouvez pas utiliser cette relation pour faire la jointure. Vous devez l'expliquer clairement".

Non, non, non, non, ce n'est pas ce que je cherchais, il doit être possible de le faire plus facilement. En lisant de plus en plus, je devenais de plus en plus déprimé et j'ai finalement dû conclure qu'il n'était pas possible de le faire plus facilement. Pire encore, pour obtenir les deux BlogPost et BlogTag d'une requête, vous devez spécifier les deux :

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

Ceci aura pour résultat d tuples 'être retourné, pas un seul objet.

En résumé, si vous avez une relation et que la plupart du temps les conditions sont différentes, il n'y a aucune utilité pour back_populates. Alors, mieux vaut oublier l'itération comme ci-dessus et commencer à penser tuples et d'autres façons d'accomplir ce que vous voulez, j'ai trouvé deux solutions possibles :

- Requête unique avec des jointures
externes - Deux requêtes SELECT-seulement, la première requête sélectionne les messages, la seconde requête sélectionne les balises pour chaque message

Ci-dessous, j'enquête sur les deux.

Solution 1 : Requête unique avec les assemblages extérieurs

La requête retourne une liste de tuples paires : post-tag. Un exemple de cette requête est :

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

Nous utilisons outerjoin parce que nous voulons que toutes les BlogPost entrées apparaissent dans le résultat. Quand un message a des balises zéro, la balise est None, quand un message a deux balises ou plus, deux balises ou plus tuples sont retournées :

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

Cela signifie que nous ne pouvons pas simplement répéter l'itération sur le tuples dans le modèle parce qu'un message peut être dans plus d'un tuple. Nous devons faire du post-traitement avant de transmettre les données au modèle. Une façon d'y parvenir est d'itérer tuples et de réduire plusieurs blogs identiques en un seul blog_post et de mettre toutes les balises d'un blog_post dans une 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)

Dans la fonction de vue, nous passons les blogs et les tags de blog au template :

def posts_list():

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

Dans le modèle, nous faisons quelque chose comme :

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

Solution 2 : Deux requêtes - SELECTseulement, la première requête sélectionnant les messages, et la seconde sélectionnant les balises pour chaque message

Pourquoi deux requêtes au lieu d'une ? N'est-ce pas toujours plus lent ? Bien que je pense que vous devriez toujours essayer de minimiser le nombre de requêtes, la différence de performance est difficile à prévoir. Certaines requêtes sont beaucoup plus rapides que d'autres. La mise en cache des requêtes peut également augmenter considérablement les performances. Pour un site Web, ne pensez jamais à la performance d'un seul visiteur, pensez toujours à ce qui se passe lorsque de nombreux visiteurs se connectent, cliquent sur un lien, etc.

La première SELECT-seule requête reçoit tous les messages :

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

La deuxième SELECT-seule requête consiste en une sélection avec toutes les conditions en place :

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

Cette requête ne renvoie tuples que les messages répondant à tous les critères, ce qui signifie qu'un message sans balises n'apparaîtra pas dans le résultat et qu'une balise ne pourra jamais être None. Comme dans la solution 1, nous effectuons le post-traitement en créant un dictionnaire avec la clé blog_post_id et une liste de valeurs contenant les tags en utilisant defaultdict. C'est maintenant beaucoup plus simple que la solution 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)

Dans la fonction de vue, nous passons les données comme nous l'avons fait dans la solution 1 :

def posts_list():

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

Le modèle ne diffère pas de la solution 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 %}

Résumé

Oubliez la belle et élégante solution mentionnée au début, elle n'existe que dans des situations très spécifiques. Les applications réelles nécessitent souvent des requêtes avec des conditions dynamiques multiples.

Avec Flask, Jinja2 et SQLAlchemy pensez-y parce tuples que ce sont les unités qui seront souvent retournées. Ceux-ci ont tuples besoin d'un post-traitement pour créer des objets adaptés à l'affichage dans le modèle.

Je n'aime pas ce post-traitement, à mon avis une requête devrait retourner toutes les données qui peuvent être mises dans un modèle, mais pour le moment je ne connais pas d'autre moyen. Je n'ai aucune préférence pour l'une ou l'autre des deux solutions ci-dessus, bien que j'ai tendance à choisir la solution à deux requêtes SELECTseulement pour sa simplicité. Pour le moment, je vais utiliser ceci.

Au cas où vous voudriez essayer ceci, voici le code complet :

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

Liens / crédits

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