Ok, a complete MCVE looks like that:
from sqlalchemy import Column, ForeignKey, Integer, String, create_engine, event
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import backref, relationship, sessionmaker
Base = declarative_base()
class Entity(Base):
__tablename__ = 'entity'
id = Column(Integer, primary_key=True)
tags = relationship('TagAssociation', backref='entity',
cascade="all, delete, delete-orphan")
class Vocabulary(Base):
__tablename__ = 'vocabulary'
id = Column(Integer, primary_key=True)
name = Column(String(255), nullable=False, unique=True)
class Tag(Base):
__tablename__ = 'tag'
id = Column(Integer, primary_key=True)
vocabularyId = Column(Integer, ForeignKey(Vocabulary.id), nullable=False)
vocabulary = relationship(Vocabulary, lazy='joined',
backref=backref('tags', cascade='all,
delete-orphan'))
name = Column(String(50), nullable=False)
class TagAssociation(Base):
__tablename__ = 'tag_association'
entityId = Column(Integer, ForeignKey(Entity.id), primary_key=True)
tagId = Column(Integer, ForeignKey(Tag.id), primary_key=True)
tag = relationship(Tag, lazy='joined')
@event.listens_for(TagAssociation, 'before_insert')
def validateOnlyOneTagPerVocabulary(mapper, connection, assoc):
vocabulary = assoc.tag.vocabulary
existingAssocs = session.query(TagAssociation).join(Tag) \
.filter(Tag.vocabulary == vocabulary, TagAssociation.entity ==
assoc.entity)
newAssocs = [assoc for assoc in assoc.entity.tags
if assoc in session.new and assoc.tag.vocabulary == vocabulary]
if len(newAssocs) > 1 or (len(newAssocs) + existingAssocs.count()) > 1:
raise ValueError('Can only assign one tag of vocabulary "%s".' %
vocabulary.name)
engine = create_engine('sqlite:///:memory:', echo=True)
Base.metadata.create_all(engine)
Session = sessionmaker(bind=engine)
session = Session()
vocabulary = Vocabulary(name='Vocabulary')
tag1 = Tag(vocabulary=vocabulary, name='Tag1')
tag2 = Tag(vocabulary=vocabulary, name='Tag2')
vocabulary2 = Vocabulary(name='Vocabulary2')
tag3 = Tag(vocabulary=vocabulary2, name='Tag3')
entity = Entity()
session.add(vocabulary)
session.add(vocabulary2)
session.add(entity)
session.commit()
TagAssociation(entity=entity, tag=tag1) # first tag
session.commit()
TagAssociation(entity=entity, tag=tag3) # tag from a different vocabulary, no
problem
session.commit()
try:
TagAssociation(entity=entity, tag=tag2) # a second tag from the same
vocabulary
session.commit()
except ValueError:
# this is expected
session.rollback()
entity.tags = [TagAssociation(entity=entity, tag=tag2)]
session.commit() # here no exception is expected
Which event or place and method are best to solve that? With „inspect“ I
couldn’t find anything that would help me.
> you'd need to show me how to illustrate the .tags collection storing two
> TagAssociation objects when you've explicitly set the collection to store a
> single TagAssociation
I don’t get it, where do I set that?
> Am 05.05.2016 um 23:26 schrieb Mike Bayer <[email protected]>:
>
>
>
> On 05/03/2016 06:28 PM, Tim-Christian Mundt wrote:
>> I have entities which I’d like to tag with tags from a vocabulary. The
>> associations between entities and tags are implemented with „manual M:N“,
>> because they carry more information.
>>
>> class Entity(Base):
>> id = Column(Integer, primary_key=True)
>> tags = relationship('TagAssociation', backref='entity',
>> cascade="all, delete, delete-orphan")
>>
>> class Vocabulary(Base):
>> id = Column(Integer, primary_key=True)
>> name = Column(String(255), nullable=False, unique=True)
>>
>> class Tag(Base):
>> id = Column(Integer, primary_key=True)
>> vocabularyId = Column(Integer, ForeignKey(Vocabulary.id), nullable=False)
>> vocabulary = relationship(Vocabulary, lazy='joined',
>> backref=backref('tags', cascade='all,
>> delete-orphan'))
>> name = Column(String(50), nullable=False)
>>
>> class TagAssociation(Base):
>> entityId = Column(Integer, ForeignKey(Entity.id), primary_key=True)
>> tagId = Column(Integer, ForeignKey(Tag.id), primary_key=True)
>> tag = relationship(Tag, lazy='joined‘)
>>
>>
>> Now I want to make sure that I don’t have more than one tag from a single
>> vocabulary on an entity. Currently I’m using
>> the before_insert event on TagAssociation which works well except in this
>> case:
>>
>> entity = Entity()
>> entity.tags.append(TagAssociation(entity=entity, tag=some_tag))
>> session.add(entity)
>> session.flush()
>> entity.tags = [TagAssociation(entity=entity,
>> tag=some_other_tag_from_same_vocabulary)]
>> session.flush()
>>
>> When the validation function is called during the second flush, the
>> collection (assoc_to_be_inserted.entity.tags)
>> has both tags in it (which I don’t understand).
>
> This is not enough detail for me to reproduce that, you'd need to show me how
> to illustrate the .tags collection storing two TagAssociation objects when
> you've explicitly set the collection to store a single TagAssociation.
> That's not anything that reproduces here.
>
>
>
> I couldn’t find a way to detect if one of them is going to be deleted.
>
> if a collection is assigned to a specific value, then the items that are no
> longer present in that collection are added to a list called "deleted". That
> can be seen like this:
>
> from sqlalchemy import inspect
>
> inspect(my_entity).attrs.tags.history
>
>
>> Hence, the validation fails, although the final result would be correct.
>>
>> Is there any event I can use to reliably implement that validation?
>
> There's not enough detail here for me to know exactly how you're doing this
> validation and under what scenario you're seeing something unexpected. The
> best format is to send an MCVE (http://stackoverflow.com/help/mcve).
>
>
>
>
>
>
>>
>> Thanks,
>> Tim
>>
>
> --
> You received this message because you are subscribed to the Google Groups
> "sqlalchemy" group.
> To unsubscribe from this group and stop receiving emails from it, send an
> email to [email protected].
> To post to this group, send email to [email protected].
> Visit this group at https://groups.google.com/group/sqlalchemy.
> For more options, visit https://groups.google.com/d/optout.
--
You received this message because you are subscribed to the Google Groups
"sqlalchemy" group.
To unsubscribe from this group and stop receiving emails from it, send an email
to [email protected].
To post to this group, send email to [email protected].
Visit this group at https://groups.google.com/group/sqlalchemy.
For more options, visit https://groups.google.com/d/optout.
