python - queries - sqlalchemy documentation

Intersección de muchos a muchos en sqlalchemy

Question

Tengo una clase de personaje con un atributo .tags; el atributo .tags es una lista de objetos Tag. en una relación de muchos a muchos. Estoy intentando escribir una consulta que encontrará todos los pares de caracteres que no tienen el mismo nombre y que tienen al menos una etiqueta en común; ¿Cómo voy a hacer esto?

Answer 1

Haga esto como sigue:

1. pensar en una consulta SQL que le dará el resultado deseado
2. crear una consulta SA correspondiente

La consulta SQL (con la cláusula WITH en SQL Server por el bien de los datos de prueba) es la siguiente (obviamente, los nombres de su tabla y columna pueden ser diferentes):

WITH t_character (id, name)
AS (    SELECT  1, "ch-1"
UNION   SELECT  2, "ch-2"
UNION   SELECT  3, "ch-3"
UNION   SELECT  4, "ch-4"
)
, t_tag (id, name)
AS (    SELECT  1, "tag-1"
UNION   SELECT  2, "tag-2"
UNION   SELECT  3, "tag-3"
)
, t_character_tag (character_id, tag_id)
AS (    SELECT  1, 1
UNION   SELECT  2, 1
UNION   SELECT  2, 2
UNION   SELECT  3, 1
UNION   SELECT  3, 2
UNION   SELECT  3, 3
UNION   SELECT  4, 3
)
-- the result should contain pairs (1, 2), (1, 3), (2, 3) again (2, 3), and (3, 4)
SELECT      DISTINCT -- will filter out duplicates
            c1.id, c2.id
FROM        t_character c1
INNER JOIN  t_character c2
        ON  c1.id < c2.id -- all pairs without duplicates
INNER JOIN  t_character_tag r1
        ON  r1.character_id = c1.id
INNER JOIN  t_character_tag r2
        ON  r2.character_id = c2.id
WHERE       r1.tag_id = r2.tag_id
ORDER BY    c1.id, c2.id

El código de muestra completo con la consulta que necesita está a continuación:

from sqlalchemy import create_engine, Column, Integer, String, ForeignKey, Table
from sqlalchemy.orm import relationship, scoped_session, sessionmaker, aliased
from sqlalchemy.ext.declarative import declarative_base

# Configure test database for SA
engine = create_engine("sqlite:///:memory:", echo=False)
session = scoped_session(sessionmaker(bind=engine, autoflush=False))

class Base(object):
    """ Just a helper base class to set properties on object creation.
    Also provides a convenient default __repr__() function, but be aware that 
    also relationships are printed, which might result in loading relations.
    """
    def __init__(self, **kwargs):
        for k,v in kwargs.items():
            setattr(self, k, v)

    def __repr__(self):
        return "<%s(%s)>" % (self.__class__.__name__, 
            ", ".join("%s=%r" % (k, self.__dict__[k]) 
                for k in sorted(self.__dict__) if "_sa_" != k[:4] and "_backref_" != k[:9])
            )
Base = declarative_base(cls=Base)

t_character_tag = Table(
    "t_character_tag", Base.metadata,
    Column("character_id", Integer, ForeignKey("t_character.id")),
    Column("tag_id", Integer, ForeignKey("t_tag.id"))
    )

class Character(Base):
    __tablename__ = u"t_character"
    id = Column(Integer, primary_key=True)
    name = Column(String)
    tags = relationship("Tag", secondary=t_character_tag, backref="characters")

class Tag(Base):
    __tablename__ = u"t_tag"
    id = Column(Integer, primary_key=True)
    name = Column(String)

# create db schema
Base.metadata.create_all(engine)


# 0. create test data
ch1 = Character(id=1, name="ch-1")
ch2 = Character(id=2, name="ch-2")
ch3 = Character(id=3, name="ch-3")
ch4 = Character(id=4, name="ch-4")
ta1 = Tag(id=1, name="tag-1")
ta2 = Tag(id=2, name="tag-2")
ta3 = Tag(id=3, name="tag-3")
ch1.tags.append(ta1)
ch2.tags.append(ta1); ch2.tags.append(ta2);
ch3.tags.append(ta1); ch3.tags.append(ta2); ch3.tags.append(ta3);
ch4.tags.append(ta3)
session.add_all((ch1, ch2, ch3, ch4,))
session.commit()

# 1. some data checks
session.expunge_all()
assert len(session.query(Character).all()) == 4
assert session.query(Tag).get(2).name == "tag-2"
assert len(session.query(Character).get(3).tags) == 3

# 2. create a final query (THE ANSWER TO THE QUESTION):
session.expunge_all()
t_c1 = aliased(Character)
t_c2 = aliased(Character)
t_t1 = aliased(Tag)
t_t2 = aliased(Tag)
q =(session.query(t_c1, t_c2).
    join((t_c2, t_c1.id < t_c2.id)).
    join((t_t1, t_c1.tags)).
    join((t_t2, t_c2.tags)).
    filter(t_t1.id == t_t2.id).
    filter(t_c1.name != t_c2.name). # if tag name is unique, this can be dropped
    order_by(t_c1.id).
    order_by(t_c2.id)
    )
q = q.distinct() # filter out duplicates
res = [_r for _r in q.all()]
assert len(res) == 4
for _r in res:
    print _r