Re: [sqlalchemy] Re: (solved) Automatic generation of changelog
Arnar Birgisson
I have this working to my liking and as promised, here's the code:
It's not complete since I'm omitting some of my model and just showing
you the relevant parts.
This is my model:
verkefni = Table("verkefni", metadata,
Column("verkefni", Integer, primary_key=True),
Column("skrad", DateTime, nullable=False, default=func.now()),
Column("sidast_breytt", DateTime, nullable=False,
default=func.current_timestamp(), onupdate=func.current_timestamp()),
Column("skrad_af", Unicode(20), ForeignKey("notendur.notandi"),
nullable=False),
Column("deadline", Date),
Column("titill", Unicode, nullable=False),
Column("lysing", Unicode, nullable=False),
Column("mikilvaegi", Integer, ForeignKey("mikilvaegi.mikilvaegi"),
nullable=False, default=40),
Column("forgangur", Integer, nullable=False, default=0),
Column("framvinda", Integer, nullable=False, default=0),
Column("fasi", Integer, ForeignKey("fasar.fasi"), nullable=False,
default=0),
Column("abyrgdarmadur", Unicode(20),
ForeignKey("notendur.notandi"), nullable=False),
Column("cc", Unicode),
Column("verknumer", Unicode(20)),
Column("tengilidur", Unicode(60)),
Column("bidur_eftir", Unicode)
)
class Verkefni(base_model):
def changelog_entry(self, notandi, skyring):
entry = Atburdur(self, notandi, skyring)
def changelog(fld, old, new):
entry.create_item(fld, old, new)
instance = self
# Scalar fields
for fld in ('deadline', 'titill', 'lysing', 'forgangur',
'framvinda', 'mikilvaegi',
'fasi', 'abyrgdarmadur', 'cc', 'verknumer',
'tengilidur', 'bidur_eftir'):
history = getattr(Verkefni, fld).get_history(instance, passive=True)
if history.is_modified():
old, new = history.deleted_items()[0], history.added_items()[0]
# Need to check this cause SA considers 10 (int) ->
10L (long) a change, we don't
if old != new:
changelog(fld, old, new)
# Fyrirtaeki
history = Verkefni.fyrirtaeki_tengsl.get_history(instance, passive=True)
for i in history.added_items():
changelog('fyrirtaeki_add', None, i.fyrirtaeki.fyrirtaeki)
for i in history.deleted_items():
changelog('fyrirtaeki_remove', i.fyrirtaeki.fyrirtaeki, None)
# Flokkar
history = Verkefni.flokkar.get_history(instance, passive=True)
for i in history.added_items():
changelog('flokkur_add', None, i.flokkur)
for i in history.deleted_items():
changelog('flokkur_remove', i.flokkur, None)
# Framkvaemd
history = Verkefni.framkvaemd.get_history(instance, passive=True)
for i in history.added_items():
changelog('framkvaemd_add', None, i.notandi.notandi)
for i in history.deleted_items():
changelog('framkvaemd_remove', i.notandi.notandi, None)
# Dependencies
history = Verkefni.depends_on.get_history(instance, passive=True)
for i in history.added_items():
changelog('dep_first_add', None, i.verkefni)
for i in history.deleted_items():
changelog('dep_first_remove', i.verkefni, None)
history = Verkefni.depend_on_me.get_history(instance, passive=True)
for i in history.added_items():
changelog('dep_then_add', None, i.verkefni)
for i in history.deleted_items():
changelog('dep_then_remove', i.verkefni, None)
# Related
history = Verkefni._relatives_a.get_history(instance, passive=False)
for i in history.added_items():
changelog('rel_add', None, i.verkefni)
for i in history.deleted_items():
changelog('rel_remove', i.verkefni, None)
history = Verkefni._relatives_b.get_history(instance, passive=False)
for i in history.added_items():
changelog('rel_add', None, i.verkefni)
for i in history.deleted_items():
changelog('rel_remove', i.verkefni, None)
atburdaskra = Table("atburdaskra", metadata,
Column("atburdur", Integer, primary_key=True),
Column("verkefni", Integer, ForeignKey("verkefni.verkefni"),
nullable=False),
Column("notandi", Unicode(20), ForeignKey("notendur.notandi"),
nullable=False),
Column("dags", DateTime, nullable=False, default=func.now()),
Column("skyring", Unicode)
)
class Atburdur(base_model):
def __init__(self, verkefni, notandi, skyring=None):
self.verkefni = verkefni
self.notandi = notandi
self.skyring = skyring
def create_item(self, svid, gamalt=None, nytt=None, texti=None):
nextid = max([0] + [item.item for item in self.items])+1
return AtburdurItem(self, nextid, svid, gamalt, nytt, texti)
assign_mapper(ctx, Atburdur, atburdaskra, properties={
'_verkefni': atburdaskra.c.verkefni,
'verkefni': relation(Verkefni, backref=backref("atburdir",
cascade="all, delete-orphan", order_by=atburdaskra.c.dags)),
'_notandi': atburdaskra.c.notandi,
'notandi': relation(Notandi)
})
atburdaskra_items = Table("atburdaskra_items", metadata,
Column("atburdur", Integer, ForeignKey("atburdaskra.atburdur"),
primary_key=True),
Column("item", Integer, primary_key=True),
Column("svid", Unicode(100)),
Column("gamalt", Unicode),
Column("nytt", Unicode),
Column("texti", Unicode)
)
def getattr_if_not_none(obj, attr, alt=""):
if obj is None:
return alt
else:
return getattr(obj, attr)
class AtburdurItem(base_model):
itemFormats = {
'flokkur_add': lambda i: u'Bætt í flokk: %s' %
getattr_if_not_none(Flokkur.get(int(i.nytt)), "heiti", i.nytt),
'fyrirtaeki_add': lambda i: u'Skráð á fyriræki: %s' %
getattr_if_not_none(Fyrirtaeki.get(int(i.nytt)), "heiti", i.nytt),
'framkvaemd_add': lambda i: u'Nýr framkvæmdaraðili: %s' %
getattr_if_not_none(Notandi.get(i.nytt), "nafn", i.nytt),
'flokkur_remove': lambda i: u'Tekið úr flokk: %s' %
getattr_if_not_none(Flokkur.get(int(i.gamalt)), "heiti", i.gamalt),
'fyrirtaeki_remove': lambda i: u'Skráð af fyriræki: %s' %
getattr_if_not_none(Fyrirtaeki.get(int(i.gamalt)), "heiti", i.gamalt),
'framkvaemd_remove': lambda i: u'Framkvæmdaraðili fjarlægður:
%s' % getattr_if_not_none(Notandi.get(i.gamalt), "nafn", i.gamalt),
'comment_add': lambda i: u'%s' % i.nytt, # TODO wiki
'forgangur': u'Forgangur',
'deadline': u'Afgr. fyrir',
'abyrgdarmadur': u'Ábyrgðarmaður',
'titill': u'Titill',
'lysing': u'Lýsing',
'cc': u'CC',
'verknumer': u'Verknúmer',
'tengilidur': u'Tengiliður',
'bidur_eftir': u'Bíður eftir',
'framvinda': u'Framvinda',
'fasi': lambda i: (u'Fasi: %s -> %s' %
(Fasi.get(i.gamalt).heiti, Fasi.get(i.nytt).heiti)),
'mikilvaegi': lambda i: (u'Mikilvægi: %s -> %s' %
(Mikilvaegi.get(i.gamalt).heiti, Mikilvaegi.get(i.nytt).heiti))
}
def __init__(self, atburdur, item, svid, gamalt=None, nytt=None,
texti=None):
self.atburdur = atburdur
self.item = item
self.svid = svid
self.gamalt = gamalt
self.nytt = nytt
self.texti = texti
def __unicode__(self):
if self.svid in AtburdurItem.itemFormats:
if callable(AtburdurItem.itemFormats[self.svid]):
return AtburdurItem.itemFormats[self.svid](self)
else:
prettyName = AtburdurItem.itemFormats[self.svid]
else:
prettyName = self.svid
def trunc(text, maxlen=70, trailer="..."):
if len(text) > maxlen:
return text[:maxlen-len(trailer)] + trailer
else:
return text
return (u'%s: %s -> %s' % (prettyName, trunc(self.gamalt),
trunc(self.nytt)))
assign_mapper(ctx, AtburdurItem, atburdaskra_items, properties={
'_atburdur': atburdaskra_items.c.atburdur,
'atburdur': relation(Atburdur, backref=backref("items",
cascade="all, delete-orphan", lazy=False))
})
Sorry for the Icelandic bits, for a crash course those are that you
need to know:
Verkefni = Project (this is the main entity of the app, which is a
project mgmt thingy)
Atburdur = Event (the changelog "entry")
AtburdurItem is a child of Atburdur and describes a transition of a field.
To use this, I just get an instance of Verkefni, change it and then
just before I flush I call changelog_entry on it with the id of the
current user and the explanation/log message he gave. This creates the
Atburdur and AtburdurItems instances according to what has changed in
the Verkefni instance.
If you don't need to specify a log message etc. one is able to take
the bulk of the changelog_entry method an place that in a after_update
on a mapper extension. Just override the nested changelog(fld, old,
new) function to insert straight to the table, log to text file or
whatever you need. Then there is no need to explicitly call anything
on the instance, just flush() the session.
Arnar