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 sqlalchemy@googlegroups.com. > To unsubscribe from this group, send email to > sqlalchemy+unsubscr...@googlegroups.com. > 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 sqlalchemy@googlegroups.com. To unsubscribe from this group, send email to sqlalchemy+unsubscr...@googlegroups.com. For more options, visit this group at http://groups.google.com/group/sqlalchemy?hl=en.