SQLalchemy 查询基于排除关系中的项目
Sqlalchemy query based on excluding items in a relationship
我有一个 table 的 PartyOrganiser(s)，一个 table 的 Contact(s) 和一个 table 的有组织的Party(s).
PartyOrganiser到Party是一对多。
PartyOrganiser 到 Contact 是 一对多 。
Party 到 Contact 是 多对多 ，具有关联 table。
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')
主要问题：
从语法上讲，对于特定聚会，我想获取与聚会组织者关联但尚未参加聚会的那些联系人的列表。 IE。称它们为 potential_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
子问题：
此配置在 PartyOrganiser、Party 和 Contact 之间具有固有的 3 向约束：各方和与会者只有在共享 party_organiser 时才能关联。 IE。 None 的 PartyOrganiser1 联系人可以参加 PartyOrganiser2 组织的 Party2。对我来说，这是否受上述格式要求的限制并不明显。事实上我相信它不是。我将如何实施此约束？
您可以在连接的 table 上使用 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()
就您的其他问题而言，您可以通过为 Party.attendees 和 Contact.intended_parties 添加属性级别验证器并确保添加到这些列表中的任何新项目来对 ORM 级别施加该约束有匹配的 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 *
我有一个 table 的 PartyOrganiser(s)，一个 table 的 Contact(s) 和一个 table 的有组织的Party(s).
PartyOrganiser到Party是一对多。
PartyOrganiser 到 Contact 是 一对多 。
Party 到 Contact 是 多对多 ，具有关联 table。
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')
主要问题：
从语法上讲，对于特定聚会，我想获取与聚会组织者关联但尚未参加聚会的那些联系人的列表。 IE。称它们为 potential_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
子问题：
此配置在 PartyOrganiser、Party 和 Contact 之间具有固有的 3 向约束：各方和与会者只有在共享 party_organiser 时才能关联。 IE。 None 的 PartyOrganiser1 联系人可以参加 PartyOrganiser2 组织的 Party2。对我来说，这是否受上述格式要求的限制并不明显。事实上我相信它不是。我将如何实施此约束？
您可以在连接的 table 上使用 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()
就您的其他问题而言，您可以通过为 Party.attendees 和 Contact.intended_parties 添加属性级别验证器并确保添加到这些列表中的任何新项目来对 ORM 级别施加该约束有匹配的 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 *