On Fri, Feb 5, 2021, at 6:37 PM, Gord Thompson wrote:
> Hi all. While trying to get more fluent with ORM I looked into a solution for
> this Stack Overflow question <https://stackoverflow.com/q/66027263/2144390>.
> For a many-to-one relationship only one child object can have
> `is_default=True`. If we add a new child object or update an existing one
> with `is_default=True` we need to set the previous "default" child object to
> `is_default=False`.
>
> I put together the following and it seems to work fine.
>
> # 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(name='{self.name}')>"
>
>
> Base.metadata.create_all(engine)
>
>
> def _remove_other_default_addrs(mapper, connection, target):
> if target.is_default:
> connection.execute(
> db.update(Address)
> .where(Address.user_id == target.user_id)
> .where(Address.id != target.id)
> .where(Address.is_default)
> .values(is_default=False)
> )
>
>
> @event.listens_for(Address, "after_insert")
> def receive_after_update(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(gord_old_addr)
> gord_new_addr = Address(user=gord, address="567 New Blvd",
> is_default=True)
> session.add(gord_new_addr)
> session.commit()
> print("==> session committed")
> print(f"==> gord_old_addr.is_default is {gord_old_addr.is_default}")
> # ==> gord_old_addr.is_default is False
>
> However, I noticed that if I flushed the session instead of committing it,
> the old child object remains unchanged
>
> print("==> session flushed")
> print(f"==> gord_old_addr.is_default is {gord_old_addr.is_default}")
> # ==> gord_old_addr.is_default is True
>
> Is there something I can add to my event listeners (or maybe the session
> configuration) to ensure that the potentially affected objects get refreshed
> without actually committing?
so this is very much like the thing that the ORM does with bulk updates, and in
fact, you might be able to use the bulk update feature to do this. the reason
being that bulk update includes the behavior where it will go into the session
and locate matching objects, and expire the attributes that were updated. to
do this, it has either the 'fetch" or "evaluate" strategy, and if you were
using 1.4 with PostgreSQL, the "fetch" strategy will actually use RETURNING to
get the matched primary key back when it does that UPDATE, the looks up that
object in the identity map by key and expires it.
You could in theory do this yourself in your _remove_other_default_attrs
method. But, it is likely safe to break one of the rules thats in the docs
regarding after_insert/after_update and actually use the ORM session version of
the method. this is much nicer in 1.4, but if you are in 1.3, you can do it
like this:
def _remove_other_default_addrs(mapper, connection, target):
if target.is_default:
session = inspect(target).session
session.query(Address).filter(
Address.user_id == target.user_id
).filter(Address.id != target.id
).filter(Address.id_default
).update(values={"is_default": False}, synchronize_session='evaluate')
now why is this "safe" do do in the mapper events? because "bulk update"
feature does not actually interact with the flush process, it just runs that
UPDATE and expires attributes that are affected.
as mentioned before I improved bulk updates a lot in 1.4, but if "evaluate"
works for the above query it should be fine in 1.3. it will actually scan
through the whole session matching that criteria on every object to find a
match.
>
> (I suppose the same question would apply if there was a server-side trigger
> on the "address" table that did the same thing. That would be my first
> choice, but it seems that lots of people are required to implement business
> logic where they cannot change the database themselves and the DBA won't do
> it for them.)
>
>
> --
> 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 sqlalchemy+unsubscr...@googlegroups.com.
> To view this discussion on the web visit
> https://groups.google.com/d/msgid/sqlalchemy/4d2990e4-78fd-48de-8e92-bd2eb39af05fn%40googlegroups.com
>
> <https://groups.google.com/d/msgid/sqlalchemy/4d2990e4-78fd-48de-8e92-bd2eb39af05fn%40googlegroups.com?utm_medium=email&utm_source=footer>.
--
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 sqlalchemy+unsubscr...@googlegroups.com.
To view this discussion on the web visit
https://groups.google.com/d/msgid/sqlalchemy/a680b80e-52fd-4d13-8958-0083b7bc9710%40www.fastmail.com.
