I have the following three models (with other columns omitted):
class ContainerSlot(db.Model):
__tablename__ = 'container_slots'
id = db.Column(db.Integer, primary_key=True)
tube_id = db.Column(db.Integer, db.ForeignKey('tubes.id'), unique=True,
index=True)
container_id = db.Column(db.Integer, db.ForeignKey('containers.id'),
nullable=False, index=True)
tube = db.relationship('Tube', uselist=False, back_populates=
'container_slot', lazy='noload')
container = db.relationship('Container', uselist=False, innerjoin=True,
back_populates='container_slots', lazy='joined')
class Container(db.Model):
__tablename__ = 'containers'
id = db.Column(db.Integer, primary_key=True)
flag = db.Column(db.Boolean)
container_slots = db.relationship('ContainerSlot', lazy='noload',
back_populates='container', order_by='ContainerSlot.id')
class Tube(db.Model):
__tablename__ = 'tubes'
id = db.Column(db.Integer, primary_key=True)
flag = db.Column(db.Boolean)
container_slot = db.relationship('ContainerSlot', uselist=False,
back_populates='tube', lazy='joined')
A container has many container slots, and a container slot has a tube. A
container slot can be "empty" (it doesn't have a tube), so its tube_id is
nullable.
I'm trying to validate that the flag is consistent between a container and
its tubes. That is, if a tube is in a container (i.e there is a container
slot with its ID as tube_id), then the tube's flag must match the value of
its container's flag.
Ideally, the only way to update these fields without failing the validation
is through endpoints which perform updates across tables in a single
transaction, ensuring that the state is consistent.
For example, I have an endpoint which sets the flag to false for a list of
tube IDs, and removes them from their container
tube_ids = args['tube_ids']
load_options = [db.noload('*'), db.joinedload(Tube.container_slot).noload(
'*')]
tubes = Tube.query.options(load_options).filter(Tube.id.in_(tube_ids)).all()
container_slots = [t.container_slot for t in tubes if t.container_slot]
for tube in tubes:
tube.flag = False
for cs in container_slots:
cs.tube_id = None
db.session.bulk_save_objects(tubes)
db.session.bulk_save_objects(container_slots)
db.session.commit()
Here is a validation which is failing
@event.listens_for(Tube, 'before_insert')
@event.listens_for(Tube, 'before_update')
def validate_tube(mapper, connection, tube):
if tube.flag is None:
return
row = (db.session.query(Container.flag, Container.name)
.join(Container.container_slots)
.join(ContainerSlot.tube)
.filter(Tube.id == tube.id)
.one_or_none())
(container_flag, name) = row if row else (None, None)
if tube.flag is True and container_flag is False:
raise BadRequest('...')
if tube.flag is False and container_flag is True:
raise BadRequest('...')
I would expect that row is None here, because the tube was removed from its
container slot, so the tube doesn't have a container
However, row returns the container that the tube was previously in, and
since the container's flag is true, the validation fails.
Some things I've already tried which didn't work are:
- Setting the container slot's tube field to None, instead of setting
tube_id to None
- Setting the tube's container_slot field to None, instead of setting
the container slot's tube or tube_id to None
- Using db.session.add instead of db.session.bulk_save_objects
I'm using SQLAlchemy 1.1.18 and Flask-SQLAlchemy 2.0.6
Thank you!
--
This e-mail is private and confidential and is for the addressee only. If
misdirected, please notify us by telephone, confirming that it has been
deleted from your system and any hard copies destroyed. You are strictly
prohibited from using, printing, distributing or disseminating it or any
information contained in it save to the intended recipient.
--
SQLAlchemy -
The Python SQL Toolkit and Object Relational Mapper
http://www.sqlalchemy.org/
To post example code, please provide an MCVE: Minimal, Complete, and Verifiable
Example. See http://stackoverflow.com/help/mcve for a full description.
---
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.
