Skip to content

Instantly share code, notes, and snippets.

@twillis
Last active December 17, 2015 01:08
Show Gist options
  • Save twillis/5525713 to your computer and use it in GitHub Desktop.
Save twillis/5525713 to your computer and use it in GitHub Desktop.
I use this snippet to call methods on the instance for before/after insert/update/delete hooks and tablename defaults to __class__.__name__
import sqlalchemy as sa
from sqlalchemy.ext.declarative import declarative_base, declared_attr
class HookEventHandler(object):
"""
registers itself with the mapper for whatever event name passed in
looks for corresponding event_name attribute on target and calls
it if it exists
"""
def __init__(self, mapper, event_name):
sa.event.listen(mapper, event_name, self)
self.event_name = event_name
def __call__(self, mapper, connection, target):
hook = getattr(target, self.event_name, None)
if hook:
return hook()
@classmethod
def register(cls):
for action in ("insert", "update", "delete"):
for occurs in ("before", "after"):
event_name = "%s_%s" % (occurs, action)
# side-effect
cls(sa.orm.mapper, event_name)
HookEventHandler.register()
#Example
class MyBase(object):
"""
base class for all models.
__tablename__ defaults to class name
"""
@declared_attr
def __tablename__(cls):
return cls.__name__
def before_insert(self):
pass # run before data is written to the database
def before_update(self):
pass # run before data is written to the database
def before_delete(self):
pass # run before data is written to the database
def after_insert(self):
pass # run before data is written to the database
def after_update(self):
pass # run before data is written to the database
def after_delete(self):
pass # run before data is written to the database
Base = declarative_base(cls=MyBase)
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment