您可以使用“after_insert”和“after_update”事件更新兄弟对象。在以下示例中,如果使用is_default=True 插入或更新地址,则事件处理程序将强制该特定用户的所有其他地址条目为is_default=False。
# https://stackoverflow.com/q/66027263/2144390
import sqlalchemy as db
from sqlalchemy import event
from sqlalchemy.orm import declarative_base, relationship
connection_uri = "sqlite://"
engine = db.create_engine(
connection_uri,
future=True,
echo=True,
)
Base = declarative_base()
class Address(Base):
__tablename__ = "address"
id = db.Column(db.Integer, primary_key=True)
user_id = db.Column(db.Integer, db.ForeignKey("user.id"), nullable=False)
address = db.Column(db.String, nullable=False)
is_default = db.Column(db.Boolean, nullable=False, default=False)
user = relationship("User", back_populates="addresses")
def __repr__(self):
return (
f"<Address(user={self.user}, address='{self.address}'"
f", is_default={self.is_default})>"
)
class User(Base):
__tablename__ = "user"
id = db.Column(db.Integer, primary_key=True)
name = db.Column(db.String)
addresses = relationship("Address", back_populates="user")
def __repr__(self):
return f"<User(id={self.id}, name='{self.name}')>"
Base.metadata.create_all(engine)
def _remove_other_default_addrs(mapper, connection, target):
if target.is_default:
session = db.inspect(target).session
session.query(Address).filter(
Address.user_id == target.user_id
).filter(Address.id != target.id).filter(
Address.is_default == True
).update(
values={"is_default": False}, synchronize_session="evaluate"
)
@event.listens_for(Address, "after_insert")
def receive_after_insert(mapper, connection, target):
_remove_other_default_addrs(mapper, connection, target)
@event.listens_for(Address, "after_update")
def receive_after_update(mapper, connection, target):
_remove_other_default_addrs(mapper, connection, target)
with db.orm.Session(engine, future=True) as session:
gord = User(name="Gord")
gord_old_addr = Address(user=gord, address="123 Old Ave", is_default=True)
session.add_all([gord, gord_old_addr])
session.commit()
print(f">>> gord_old_addr.is_default is {gord_old_addr.is_default}")
# >>> gord_old_addr.is_default is True
gord_new_addr = Address(user=gord, address="567 New Blvd", is_default=True)
session.add(gord_new_addr)
session.flush()
print(">>> session flushed")
print(f">>> gord_old_addr.is_default is {gord_old_addr.is_default}")
# >>> gord_old_addr.is_default is False