listen to deleted_to_detached event trough zope

test_deleted_to_detached_zope.py

# from sqlalchemy import Integer, ForeignKey, String, Column, create_engine
# from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import relationship, Session
from sqlalchemy import event

from sqlalchemy import *
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import scoped_session, sessionmaker, relation
from zope.sqlalchemy import ZopeTransactionExtension, register
import transaction

Base = declarative_base()


class User(Base):
    __tablename__ = 'user'
    id = Column(Integer, primary_key=True)
    name = Column(String)

    addresses = relationship("Address", back_populates="user")

class Address(Base):
    __tablename__ = 'address'
    id = Column(Integer, primary_key=True)
    email = Column(String)
    user_id = Column(Integer, ForeignKey('user.id'))

    user = relationship("User", back_populates="addresses")


def test_event(session, instance):
    print("do something here after delete has been committed")


if __name__ == '__main__':
    e = create_engine("sqlite://", echo=True)
    Base.metadata.create_all(e)

    Session = scoped_session(sessionmaker(
        bind=e
    ))

    register(Session)

    s = Session()
    event.listen(s, 'deleted_to_detached', test_event)

    s.add(User(id=1))
    s.add(Address(id=1, user_id=1))
    transaction.commit()

    u1 = s.query(User).get(1)
    s.delete(u1)
    transaction.commit()
    s.close()