I have a SQLAlchemy entity that I’d like to track the “before” and “after” states on in an event capture. e.g. user.name was “robert”, and then it was updated to “bob”. In the example below, I only have access to “bob” (via sqlalchemy’s get_history) at the time of the event being triggered.
More concretely, I have this entity:
<code>class Order(Base):
user_id = Column(Integer, ForeignKey("users.id"), nullable=False, index=True)
name = Column(String, nullable=True)
def __init__(
self,
user_id=None,
name=None,
):
self.user_id = user_id
self.name = name
</code>
<code>class Order(Base):
user_id = Column(Integer, ForeignKey("users.id"), nullable=False, index=True)
name = Column(String, nullable=True)
def __init__(
self,
user_id=None,
name=None,
):
self.user_id = user_id
self.name = name
</code>
class Order(Base):
user_id = Column(Integer, ForeignKey("users.id"), nullable=False, index=True)
name = Column(String, nullable=True)
def __init__(
self,
user_id=None,
name=None,
):
self.user_id = user_id
self.name = name
With this event listener:
<code>def handle_change(_mapper, _connection, target):
added_name, _, deleted_name = get_history(target, "deleted_on")
added_user_id, _, deleted_user_id = get_history(target, "user_id")
if len(added_name) > 0:
print(f"Added a name {added_name[0]}")
if len(deleted_name) > 0:
print(f"Deleted a name {deleted_name[0]}")
if len(added_user_id) > 0:
print(f"Added a user_id {added_user_id[0]}")
if len(deleted_user_id) > 0:
print(f"Deleted a user_id {deleted_user_id[0]}")
event.listen(
Order,
"after_insert",
handle_change,
)
event.listen(
Order,
"after_update",
handle_change,
)
</code>
<code>def handle_change(_mapper, _connection, target):
added_name, _, deleted_name = get_history(target, "deleted_on")
added_user_id, _, deleted_user_id = get_history(target, "user_id")
if len(added_name) > 0:
print(f"Added a name {added_name[0]}")
if len(deleted_name) > 0:
print(f"Deleted a name {deleted_name[0]}")
if len(added_user_id) > 0:
print(f"Added a user_id {added_user_id[0]}")
if len(deleted_user_id) > 0:
print(f"Deleted a user_id {deleted_user_id[0]}")
event.listen(
Order,
"after_insert",
handle_change,
)
event.listen(
Order,
"after_update",
handle_change,
)
</code>
def handle_change(_mapper, _connection, target):
added_name, _, deleted_name = get_history(target, "deleted_on")
added_user_id, _, deleted_user_id = get_history(target, "user_id")
if len(added_name) > 0:
print(f"Added a name {added_name[0]}")
if len(deleted_name) > 0:
print(f"Deleted a name {deleted_name[0]}")
if len(added_user_id) > 0:
print(f"Added a user_id {added_user_id[0]}")
if len(deleted_user_id) > 0:
print(f"Deleted a user_id {deleted_user_id[0]}")
event.listen(
Order,
"after_insert",
handle_change,
)
event.listen(
Order,
"after_update",
handle_change,
)
And this test
<code># create_user is a fixture to create a valid user
associated_user = create_user() # id = 1
order = Order()
order.name = "Testing Order"
order.user_id = associated_user.id
order.save()
# Added a name Testing Order
# Added a user_id 1
new_user = create_user() # id = 2
order.name = "Updated"
order.user_id = new_user.id
order.save()
# Added a name Updated
# Added a user_id 2
# I want there to also be:
# Deleted a name Testing Order
# Deleted a user_id 1
</code>
<code># create_user is a fixture to create a valid user
associated_user = create_user() # id = 1
order = Order()
order.name = "Testing Order"
order.user_id = associated_user.id
order.save()
# Added a name Testing Order
# Added a user_id 1
new_user = create_user() # id = 2
order.name = "Updated"
order.user_id = new_user.id
order.save()
# Added a name Updated
# Added a user_id 2
# I want there to also be:
# Deleted a name Testing Order
# Deleted a user_id 1
</code>
# create_user is a fixture to create a valid user
associated_user = create_user() # id = 1
order = Order()
order.name = "Testing Order"
order.user_id = associated_user.id
order.save()
# Added a name Testing Order
# Added a user_id 1
new_user = create_user() # id = 2
order.name = "Updated"
order.user_id = new_user.id
order.save()
# Added a name Updated
# Added a user_id 2
# I want there to also be:
# Deleted a name Testing Order
# Deleted a user_id 1