On Sep 20, 2011, at 8:33 AM, Andrei Chirila wrote:
> Hello,
>
> I'm working at implementing a safe delete on my models (having a
> deleted_at field which is checked and setting a deleted_at field when
> a model is deleted).
>
> I looked over the LimitingQuery recipe and at
> SessionEvents.before_flush and bellow is my solution. As I'm kind of
> new SqlAlchemy user, I would like to ask if anyone can see any obvious
> mistake in the following code (I'm sure I missed bulk deletion and
> session explunge, but I don't really know how to integrate them).
I'd stay away from identity_map.add(object), as that skips a lot of bookkeeping
that occurs within the Session. and definitely *never* modify anything with
an _underscore on Session, your code becomes totally non portable to even small
changes in SQLAlchemy (which occur frequently as we have new releases, bugfixes
you'll want, etc). The public (read-only) accessor for the "deleted"
collection is Session.deleted.
If you have an object marked for deletion, you can re-add it,
session.add(object), will remove it from the deleted list.
If there's something you need the Session to do that there is not a public API
for, ask me on the list here first, as I'd rather ensure there are maintainable
paths for use cases.
>
> Any feedback would be appreciated.
>
> Thank you,
> Andrei Chirila
>
> from sqlalchemy.orm.query import Query
> import sqlalchemy.event as event
>
>
> class DeletedAtQuery(Query):
> """ Query adding a default condition that,
> if deleted_at exists, it should be NULL
> for the rows returned """
>
> def get(self, ident):
> return Query.get(self.populate_existing(), ident)
>
> def __iter__(self):
> return Query.__iter__(self.private())
>
> def from_self(self, *ent):
> return Query.from_self(self.private(), *ent)
>
> def private(self):
> mzero = self._mapper_zero()
> if mzero is not None:
> if hasattr(mzero.class_, 'deleted_at'):
> crit = mzero.class_.deleted_at == None
>
> return self.filter(crit)
> return self
>
>
> def paranoid_delete(session, flush_context, instances):
> """ Intercepts item deletions (single item deletions)
> and, if there is a field deleted_at for that
> object, it's updating deleted at instead of really
> deleting the object """
> # print "current session", session
> # print "context", flush_context
> # print "dirty", session.dirty
> # print "deleted", session._deleted
> deleted_copy = set(session._deleted)
> for instance in deleted_copy:
> obj = instance.obj()
> # print ">> deleted object:", instance, " -> ", obj
> if hasattr(obj, 'deleted_at'):
> # print "moving instance to dirty states"
> instance.deleted = False
> obj.deleted_at = int(time.time())
> session.identity_map.add(instance)
> del session._deleted[instance]
>
>
> def configure_events(session):
> """ Helper for applying before_flush hook """
> event.listen(session, 'before_flush', paranoid_delete)
>
> if __name__ == "__main__":
>
> import time
> from sqlalchemy import *
> from sqlalchemy.orm import *
> from sqlalchemy.ext.declarative import declarative_base
>
> Base = declarative_base()
>
> class Group(Base):
> __tablename__ = 'groups'
>
> id = Column(Integer, primary_key=True)
> name = Column(String)
> users = relation("User", backref="group", cascade="all")
>
> def __repr__(self):
> return "<Group(%d, %s)>" % (self.id, self.name)
>
> class User(Base):
> __tablename__ = 'users'
>
> id = Column(Integer, primary_key=True)
> name = Column(String)
> group_id = Column(Integer, ForeignKey('groups.id'))
> deleted_at = Column(Integer)
> pets = relation("Pet", backref="user", cascade="all, delete-
> orphan")
>
> def __repr__(self):
> return "<User(%d, %s)>" % (self.id, self.name)
>
> class Pet(Base):
> __tablename__ = 'pets'
>
> id = Column(Integer, primary_key=True)
> name = Column(String)
> user_id = Column(Integer, ForeignKey('users.id'),
> nullable=False)
> deleted_at = Column(Integer)
>
> def __repr__(self):
> return "<Pet(%d, %s)>" % (self.id, self.name)
>
> engine = create_engine("sqlite://", echo=False)
> Base.metadata.create_all(engine)
>
> ClearSession = sessionmaker(bind=engine)
> Session = sessionmaker(bind=engine, query_cls=DeletedAtQuery)
>
> sess = Session()
> configure_events(sess)
> clearsess = ClearSession()
>
> sess.add_all([Group(name='Group A', users=[
> User(name='User A1', pets=[
> Pet(name='Mause A1'),
> Pet(name='Cat A1')])]),
> Group(name='Group B', users=[
> User(name='User B1'),
> User(name='User B2', pets=[
> Pet(name='Dog B2'),
> Pet(name='Mause B2'),
> Pet(name='Hamster B2')
> ]),
> User(name='User B3', pets=[
> Pet(name='Fish B3'),
> Pet(name='Snake B3'),
> Pet(name='Horse B3')
> ])])])
>
> sess.commit()
>
> group_a = sess.query(Group).filter(Group.name == 'Group
> A').first()
> assert group_a
>
> users_in_a = group_a.users
> assert users_in_a
> assert len(users_in_a) == 1
> users_in_a_by_query = sess.query(User).\
> filter(User.group_id == group_a.id).\
> all()
>
> assert users_in_a_by_query
> assert len(users_in_a_by_query) == 1
>
> pets_in_a = users_in_a[0].pets
> assert pets_in_a
> assert len(pets_in_a) == 2
>
> # delete one pet
> pet_id = pets_in_a[0].id
> sess.delete(pets_in_a[0])
> sess.commit()
>
> pets_in_a = users_in_a[0].pets
> assert pets_in_a
> assert len(pets_in_a) == 1
>
> pet = clearsess.query(Pet).filter(Pet.id == pet_id).first()
> assert pet
> assert pet.deleted_at
>
> sess.delete(users_in_a[0])
> sess.commit()
>
> assert len(sess.query(User).\
> filter(User.group_id == group_a.id).\
> all()) == 0
> assert len(clearsess.query(User).\
> filter(User.group_id == group_a.id).\
> all()) == 1
> assert len(sess.query(Pet).all()) == 6
> assert len(clearsess.query(Pet).all()) == 8
>
> group_b = sess.query(Group).filter(Group.name == 'Group
> B').first()
> assert group_b
> assert len(group_b.users) == 3
>
> # don't know how to implement batch delete hook yet
> #sess.query(Pet).delete()
> #sess.commit()
>
> #assert len(sess.query(Pet).all()) == 0
> #assert len(clearsess.query(Pet).all()) == 8
>
> sess.delete(group_b)
> sess.commit()
> groub_b_deleted = sess.query(Group).filter(Group.name == 'Group
> B').first()
> assert groub_b_deleted is None
>
> assert len(sess.query(Pet).all()) == 0
> assert len(clearsess.query(Pet).all()) == 8
> assert len(sess.query(User).all()) == 0
> assert len(clearsess.query(User).all()) == 4
>
> --
> You received this message because you are subscribed to the Google Groups
> "sqlalchemy" group.
> To post to this group, send email to [email protected].
> To unsubscribe from this group, send email to
> [email protected].
> For more options, visit this group at
> http://groups.google.com/group/sqlalchemy?hl=en.
>
--
You received this message because you are subscribed to the Google Groups
"sqlalchemy" group.
To post to this group, send email to [email protected].
To unsubscribe from this group, send email to
[email protected].
For more options, visit this group at
http://groups.google.com/group/sqlalchemy?hl=en.
