Sqlalchemy запрос, основанный на исключении элементов в отношениях

Question

У меня есть таблица PartyOrganiser (s), таблица своих Contact (s) и таблица организованных Party (s).

PartyOrganiser to Party - это один-ко-многим .
PartyOrganiser до Contact - это один-ко-многим .
Party до Contact - это «многие ко многим» с таблицей ассоциации.

class PartyOrganiser(db.Model):
    __tablename__ = 'party_organiser'
    id   = db.Column(db.Integer, primary_key=True)
    name = db.Column(db.String)
    parties  = db.relationship("Party",    back_populates='party_organiser')
    contacts = db.relationship("Contacts", back_populates='party_organiser')

contact_party_ass_tab = db.Table('contact_party', db.Model.metadata,
    db.Column('party_id', db.Integer, db.ForeignKey('party.id')),
    db.Column('contact_id', db.Integer, db.ForeignKey('contact.id')))

class Party(db.Model):
    __tablename__ = 'party'
    id      = db.Column(db.Integer, primary_key=True)
    details = db.Column(db.String)
    party_organiser_id = db.Column(db.Integer, db.ForeignKey('party_organiser.id'), nullable=False)
    party_organiser = db.relationship("PartyOrganiser", back_populates="parties", uselist=False)
    attendees = db.relationship("Contact", secondary=contact_party_ass_tab, back_populates='intended_parties')

class Contact(db.Model):
    __tablename__ = 'contact'
    id   = db.Column(db.Integer, primary_key=True)
    name = db.Column(db.String)
    party_organiser_id = db.Column(db.Integer, db.ForeignKey('party_organiser.id'), nullable=False)
    party_organiser = db.relationship("PartyOrganiser", back_populates="parties", uselist=False)
    intended_parties = db.relationship("Contact", secondary=contact_party_ass_tab, back_populates='attendees')

Основной вопрос:

Грамматически для конкретной партии я хочу получить список тех контактов, связанных с организатором вечеринки, которые еще не посещают вечеринку.То есть, называя их потенциальными_аттени, я хотел бы следующее как решение стиля запроса SQLalchemy:

class Party(db.model):
    ...
    @property
    def potential_attendees(self):
        # get all contacts for the right party_organiser
        sub_query = Contact.query.filter(Contact.party_organiser_id == self.party_organiser_id)
        # then eliminate those that are already attendees to this party..
        return sub_query.difference(self.attendees) # <- pseudocode

Под вопрос:

Эта конфигурация имеет 3-х стороннее ограничение между PartyOrganiser, Party и Contact: стороны и участники могут быть связаны, только если они совместно используют party_organiser.Т.е. ни один из контактов PartyOrganiser1 не может быть участником Party2, организованной PartyOrganiser2.Для меня не очевидно, что это ограничено, как требуется в вышеуказанном формате.На самом деле я верю, что это не так.Как бы я реализовал это ограничение?

Answer 1

Вы можете запросить элементы, которые исключены в связи, используя конструкцию NOT EXISTS в объединенной таблице.

@property
def potential_attendees(self):
    sq = db.session.query(contact_party_ass_tab.columns.contact_id).subquery()
    return db.session.query(Contact).filter(
            Contact.party_organiser_id==self.party_organiser_id,
            ~exists().where(sq.c.contact_id==Contact.id)
        ).all()

Что касается вашего другого вопроса, вы можете наложить это ограничение на уровень ORM, добавив валидаторы уровня атрибута для Party.attendees и Contact.intended_parties и убедившись, что любой новый элемент добавлен в эти списки.имеет соответствующий party_organiser_id.Вот полный код

from flask import Flask
from flask.ext.sqlalchemy import SQLAlchemy
from sqlalchemy import exists

from sqlalchemy.orm import validates


app = Flask(__name__)
app.config['SQLALCHEMY_DATABASE_URI'] = 'sqlite:///:memory:'
db = SQLAlchemy(app)

class PartyOrganiser(db.Model):
    __tablename__ = 'party_organiser'
    id   = db.Column(db.Integer, primary_key=True)
    name = db.Column(db.String)
    parties  = db.relationship("Party",    back_populates='party_organiser')
    contacts = db.relationship("Contact", back_populates='party_organiser')

    def __repr__(self):
        return self.name


contact_party_ass_tab = db.Table('contact_party', db.Model.metadata,
    db.Column('party_id', db.Integer, db.ForeignKey('party.id')),
    db.Column('contact_id', db.Integer, db.ForeignKey('contact.id')))

class Party(db.Model):
    __tablename__ = 'party'
    id      = db.Column(db.Integer, primary_key=True)
    name = db.Column(db.String)
    details = db.Column(db.String)
    party_organiser_id = db.Column(db.Integer, db.ForeignKey('party_organiser.id'), nullable=False)
    party_organiser = db.relationship("PartyOrganiser", back_populates="parties", uselist=False)
    attendees = db.relationship("Contact", secondary=contact_party_ass_tab, back_populates='intended_parties')

    def __repr__(self):
        return self.name

    @property
    def potential_attendees(self):
        sq = db.session.query(contact_party_ass_tab.columns.contact_id).subquery()
        return db.session.query(Contact).filter(
                Contact.party_organiser_id==self.party_organiser_id,
                ~exists().where(sq.c.contact_id==Contact.id)
            ).all()

    @validates('attendees')
    def validate_attendee(self, key, attendee):
        assert attendee.party_organiser_id == self.party_organiser_id
        return attendee

class Contact(db.Model):
    __tablename__ = 'contact'
    id   = db.Column(db.Integer, primary_key=True)
    name = db.Column(db.String)
    party_organiser_id = db.Column(db.Integer, db.ForeignKey('party_organiser.id'), nullable=False)
    party_organiser = db.relationship("PartyOrganiser", back_populates="contacts", uselist=False)
    intended_parties = db.relationship("Party", secondary=contact_party_ass_tab, back_populates='attendees')

    def __repr__(self):
        return self.name

    @validates('intended_parties')
    def validate_party(self, key, party):
        assert party.party_organiser_id == self.party_organiser_id
        return party

db.create_all()

organiser1 = PartyOrganiser(name="organiser1")
organiser2 = PartyOrganiser(name="organiser2")

db.session.add_all([organiser1, organiser2])
db.session.commit()


org1_party1 = Party(name="Organiser1's Party1", party_organiser_id=organiser1.id)
org1_party2 = Party(name="Organiser1's Party2", party_organiser_id=organiser1.id)

org2_party1 = Party(name="Organiser2's Party1", party_organiser_id=organiser2.id)
org2_party2 = Party(name="Organiser2's Party2", party_organiser_id=organiser2.id)

db.session.add_all([org1_party1, org1_party2, org2_party1, org2_party2])
db.session.commit()

org1_contact1 = Contact(name="Organiser1's contact 1", party_organiser_id=organiser1.id)
org1_contact2 = Contact(name="Organiser1's contact 2", party_organiser_id=organiser1.id)
org1_contact3 = Contact(name="Organiser1's contact 3", party_organiser_id=organiser1.id)
org1_contact4 = Contact(name="Organiser1's contact 4", party_organiser_id=organiser1.id)

org2_contact1 = Contact(name="Organiser2's contact 1", party_organiser_id=organiser2.id)
org2_contact2 = Contact(name="Organiser2's contact 2", party_organiser_id=organiser2.id)
org2_contact3 = Contact(name="Organiser2's contact 3", party_organiser_id=organiser2.id)
org2_contact4 = Contact(name="Organiser2's contact 4", party_organiser_id=organiser2.id)

db.session.add_all([org1_contact1, org1_contact2, org1_contact3, org1_contact4, org2_contact1, org2_contact2, org2_contact3, org2_contact4])
db.session.commit()

org1_party1.attendees.append(org1_contact1)
db.session.commit()

print "Potential attendees of org1_party1 ", org1_party1.potential_attendees

print "Attempting to add a contact of a different organiser. Will throw exception"

org1_party1.attendees.append(org2_contact1)

Вывод (проверьте исключение в конце, которое выдается последней строкой в ​​приведенном выше коде):

In [1]: from exclusion_query import *
/home/surya/Envs/inkmonk/local/lib/python2.7/site-packages/flask_sqlalchemy/__init__.py:794: FSADeprecationWarning: SQLALCHEMY_TRACK_MODIFICATIONS adds significant overhead and will be disabled by default in the future.  Set it to True or False to suppress this warning.
  'SQLALCHEMY_TRACK_MODIFICATIONS adds significant overhead and '
Potential attendees of org1_party1  [Organiser1's contact 2, Organiser1's contact 3, Organiser1's contact 4]
Attempting to add a contact of a different organiser. Will throw exception
---------------------------------------------------------------------------
AssertionError                            Traceback (most recent call last)
<ipython-input-1-4380704ace46> in <module>()
----> 1 from exclusion_query import *