Any comment is welcome as I don't really know SA internals, it seems to
work against a simple mapped table.
#------------------------------------------------------------------
import sqlalchemy as sqa
import sqlalchemy.orm as sqorm
from sqlalchemy.sql import text as sqaText
class AuditLog(sqorm.MapperExtension):
def __init__(self, table_prefix):
self.table_prefix = table_prefix
def after_update(self, mapper, connection, instance):
#what to do with a multitable mapper ?
assert len(mapper.tables) == 1
statement = "INSERT INTO %s%s" %(self.table_prefix,
mapper.tables[0].name)
fnames = ["logoperation"]
parms = [":logoperation"]
vals = {"logoperation" : "U"} #the log operation is U = Update
for c in mapper.columns:
attrName = c.name
fnames.append(attrName)
parm = ":%s" % attrName
parms.append(parm)
(added, unchanged, deleted) = sqorm.attributes.get_history(
instance._state,
attrName)
#when it fails?
assert len(unchanged) < 2
if added :
vals[attrName] = added[0]
elif unchanged:
vals[attrName] = unchanged[0]
fnamesStr = ", ".join(fnames)
valuesStr = ", ".join(parms)
statement += " (%s) VALUES (%s)" % (fnamesStr, valuesStr)
s=sqaText(statement)
connection.execute(s, vals)
return sqorm.EXT_CONTINUE