Flask Jinja2 y SLQAlchemy many-to-many relación con las condiciones

Descargo de responsabilidad: Este es mi primer SQLAlchemy proyecto, por lo que no tengo experiencia y puedo hacer declaraciones erróneas en este post.

Mientras trabajaba en mi primer Flask proyecto de SQLAchemy (no uso Flask-SQLAlchemy) me golpeé con una pared muy alta y gruesa. Para explicar esto, considera que tienes un blog con entradas y etiquetas. Existe una many-to-many relación entre los mensajes y las etiquetas: un mensaje puede tener muchas etiquetas y una etiqueta puede tener muchos mensajes.

La tabla de asociación y los modelos:

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) 

Fácil, agradable, sin problemas. En la función de vista pasamos los mensajes a la plantilla:

def posts_list():

    blog_posts = db.query(BlogPost)

    return render_template(
        ...
        blog_posts=blog_posts)

En la plantilla Ninja iteramos sobre los mensajes y etiquetas:

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

¡Hermoso, elegante! ¿Pero realista?

El problema: las condiciones

Entre las SQLAlchemy preguntas más vistas, por ejemplo, sobre el stackoverflow, se encuentra la many-to-many relación con las condiciones. Por qué? Porque empezar por ahí SQLAlchemy es confuso.

Te encuentras con problemas cuando quieres añadir una o más condiciones. ¿Cómo se hace esto? La relación de ordenación definida ya no puede utilizarse o puede utilizarse? Creía que habría una solución sencilla para añadir, por ejemplo, una condición de borrado o un filtro en el nombre de la etiqueta. Mientras buscaba soluciones, encontré este texto, vea el primer enlace a continuación: Si necesita una condición JOIN diferente a la que define la relación(), no puede usar esa relación para hacer la unión. Hay que explicarlo con claridad".

No, no, no, esto no era lo que estaba buscando, debe ser posible hacerlo más fácil. Leyendo más y más me deprimí más y más y finalmente tuve que concluir que no era posible hacer esto más fácil. Peor aún, para obtener ambos BlogPost y BlogTag de una consulta, debe especificar ambos:

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

Esto resultará en tuples ser devuelto, ni un solo objeto.

Resumiendo, si usted tiene una relación y la mayoría de las veces las condiciones son diferentes, entonces no hay uso para back_populates. Así que mejor olvídate de la iteración como la anterior y empieza a pensar tuples y otras formas de lograr lo que quieres, se me ocurrieron dos posibles soluciones:

- Consulta simple con uniones
externas - Dos consultas SELECT-sólo, primera consulta seleccionando los mensajes, segunda consulta seleccionando etiquetas para cada mensaje.

A continuación investigo ambos.

Solución 1: consulta única con uniones externas

La consulta devuelve una lista de tuples pares: post-tag. Un ejemplo de esta consulta es:

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

Usamos outerjoin porque queremos que todas las BlogPost entradas aparezcan en el resultado. Cuando un mensaje tiene cero etiquetas, la etiqueta es None, cuando un mensaje tiene dos o más etiquetas, se devuelven dos o más tuples :

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

Esto significa que no podemos simplemente iterar sobre el tuples en la plantilla porque un mensaje puede estar en más de uno tuple. Debemos hacer algún post-procesamiento antes de pasar los datos a la plantilla. Una forma de hacer esto es iterar tuples y reducir múltiples blogs_posts idénticos a un solo blog_post y poner todas las etiquetas para un blog_post en una lista:

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)

En la función ver pasamos los blog_posts y blog_tags a la plantilla:

def posts_list():

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

En la plantilla hacemos algo como:

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

Solución 2: Dos consultas SELECT, la primera seleccionando los mensajes y la segunda seleccionando etiquetas para cada mensaje.

¿Por qué dos consultas en lugar de una? ¿No es siempre más lento? Aunque creo que siempre se debe tratar de minimizar el número de consultas, la diferencia de rendimiento es difícil de predecir. Algunas consultas son mucho más rápidas que otras. El almacenamiento en caché de consultas también puede aumentar drásticamente el rendimiento. Para un sitio web, nunca piense en el rendimiento de un solo visitante, siempre considere lo que sucede cuando muchos visitantes se conectan, hacen clic en un enlace, etc.

La primera SELECTconsulta es la única que recibe todos los mensajes:

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

La segunda SELECTconsulta consiste en una selección con todas las condiciones establecidas:

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

Esta consulta sólo devuelve tuples los resultados que cumplen todos los criterios, lo que significa que un mensaje sin etiquetas no aparecerá en el resultado y que una etiqueta nunca podrá ser None... Como en la solución 1, hacemos el post-procesamiento creando un diccionario con la clave blog_post_id y un valor de una lista que contiene las etiquetas usando defaulttdict. Esto es ahora mucho más sencillo que la solución 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)

En la función de vista pasamos los datos como hicimos en la solución 1:

def posts_list():

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

La plantilla no difiere de la solución 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 %}

Resumen

Olvídate de la bella y elegante solución mencionada al principio, sólo existe en situaciones muy específicas. Las aplicaciones de la vida real a menudo requieren consultas con múltiples condiciones dinámicas.

Con Flask, Jinja2 y SQLAlchemy piensa en tuples porque estas son las unidades que a menudo serán devueltas. Éstos tuples necesitan un tratamiento posterior para crear objetos que puedan visualizarse en el modelo.

No me gusta este post-procesamiento, en mi opinión una consulta debería devolver todos los datos que se pueden poner en una plantilla, pero por el momento no conozco otra forma. No tengo preferencia por ninguna de las dos soluciones anteriores, aunque tiendo a elegir la solución de sólo dos consultas SELECTpor su simplicidad. Por el momento voy a usar esto.

En caso de que quieras probar esto, aquí está el código completo:

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

Enlaces / créditos

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