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).
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.
