On 10/29/07, mmstud <[EMAIL PROTECTED]> wrote:
>
> On 29 loka, 09:08, [EMAIL PROTECTED] wrote:
> > > I dont need history tracking, just revert documents to older ones.
> >
> > that is history, just not timed history.
>
> Most optimal would it be, if i can make rows with updated fields only,
> not to copy whole row... but im afraid setting unchanged field to None
> would be problematic when retrieving versions. I should retrieve
> several rows and collect the latest not None fields... just pondering
That reminded me, I have another project that does automatic
changelogs. There is something in the ML archives, but here's some
more code.
I'm afraid all class, variable and field names are in Icelandic - I
put in comments with translations where it matters. I also removed
alot of auxiliary tables since this is proprietary code. The base
entity that keeps a changelog of itself is Verkefni (means "project").
A changelog entry is generated by calling changelog_entry on a dirty
object. This is not done automatically as sometimes one wants to
update an object without generating a log entry.
Arnar
# -*- encoding: UTF-8 -*-
import datetime
import cherrypy
import sqlalchemy
from sqlalchemy.ext.sessioncontext import SessionContext
from sqlalchemy.ext.assignmapper import assign_mapper
from sqlalchemy.orm import MapperExtension, EXT_PASS
_engine = None
metadata = sqlalchemy.DynamicMetaData()
def get_engine():
global _engine
dburi = cherrypy.config.get('verkstjorinn.db.uri')
encoding = cherrypy.config.get('verkstjorinn.db.encoding', 'utf-8')
echo = cherrypy.config.get('verkstjorinn.db.echo', False)
if not _engine:
_engine = sqlalchemy.create_engine(dburi, encoding=encoding,
echo=echo, convert_unicode=True)
metadata.connect(_engine)
elif not metadata.is_bound():
metadata.connect(_engine)
return _engine
ctx = SessionContext(lambda:sqlalchemy.create_session(bind_to=get_engine()))
def sa_uow_cleanup():
ctx.current.clear()
sa_uow_cleanup.failsafe = True
cherrypy.tools.sacleanup = cherrypy.Tool('on_end_request', sa_uow_cleanup)
class base_model(object):
def __repr__(self):
props = []
for key in self.c.keys():
props.append("%s=%r" % (key, getattr(self, key)))
return "<%s %s>" % (self.__class__.__name__, ' '.join(props))
from sqlalchemy import *
## Verkefni = Projects
verkefni_flokkar = Table("verkefni_flokkar", metadata,
Column("verkefni", Integer, ForeignKey("verkefni.verkefni"),
primary_key=True),
Column("flokkur", Integer, ForeignKey("flokkar.flokkur"), primary_key=True)
)
# No mapper for this table, it's only a join table for the many-to-many relation
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=5),
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):
class DEFAULTS:
def __init__(self):
self.verkefni = 0
self.skrad = datetime.datetime.now()
self.sidast_breytt = datetime.datetime.now()
self.titill = ""
self.lysing = ""
self.verknumer = None
self.tengilidur = None
self.deadline = None
self.bidur_eftir = None
self.flokkar = []
self.mikilvaegi = Mikilvaegi.get(40)
self.forgangur = 0
self.framvinda = 0
self.fasi = Fasi.get(5)
self.abyrgdarmadur = None # Must be set by controller
self.cc = None
self.depends_on = ()
self.depend_on_me = ()
def get_fyrirtaeki(self): return []
def get_framkvaemd(self): return []
def combined_athugasemdir(self): return []
def get_dependency_parents(self): return []
def get_dependency_children(self): return []
@property
def skyldverkefni(self):
related = SkyltVerkefni.select_by(
or_(SkyltVerkefni.c._verkefni_a==self.verkefni,SkyltVerkefni.c._verkefni_b==self.verkefni)
)
relations = []
for sv in related:
if sv.verkefni_a == self:
relations.append((sv.verkefni_b, sv.athugasemd))
else:
relations.append((sv.verkefni_a, sv.athugasemd))
return relations
def combined_athugasemdir(self):
l = list(self.textar) + [a for a in self.atburdir if a.skyring
!= u'Bætti við athugasemd.']
l.sort(key=lambda x: x.dags)
return l
def get_fyrirtaeki(self):
return [a.fyrirtaeki for a in self.fyrirtaeki_tengsl]
def add_fyrirtaeki(self, fyrirtaeki_to_add):
va = VerkefniFyrirtaekiAssociation()
va.fyrirtaeki = fyrirtaeki_to_add
va.verkefni = self
def remove_fyrirtaeki(self, fyrirtaeki_to_remove):
for va in list(self.fyrirtaeki_tengsl):
if va.fyrirtaeki == fyrirtaeki_to_remove:
self.fyrirtaeki_tengsl.remove(va)
def get_framkvaemd(self):
return [a.notandi for a in self.framkvaemd]
def add_framkvaemd(self, notandi_to_add):
f = Framkvaemd()
f.notandi = notandi_to_add
f.verkefni = self
def remove_framkvaemd(self, notandi_to_remove):
for f in list(self.framkvaemd):
if f.notandi == notandi_to_remove:
self.framkvaemd.remove(f)
def get_dependency_children(self):
return [d.child for d in self.depend_on_me]
def add_dependency_child(self, child):
if child not in self.get_dependency_children():
d = Dependency()
d.parent = self
d.child = child
self.depend_on_me.append(d)
def remove_dependency_child(self, child):
dd = [d for d in self.depend_on_me if d.child == child]
for d in dd:
self.depend_on_me.remove(d)
def get_dependency_parents(self):
return [d.parent for d in self.depends_on]
def add_dependency_parent(self, parent):
if parent not in self.get_dependency_parents():
d = Dependency()
d.parent = parent
d.child = self
self.depends_on.append(d)
def remove_dependency_parent(self, parent):
dd = [d for d in self.depends_on if d.parent == parent]
for d in dd:
self.depends_on.remove(d)
def add_comment(self, notandi, texti):
t = Texti()
t.dags = datetime.datetime.now()
t.notandi = notandi
t.body = texti
self.textar.append(t)
a = Atburdur(self, notandi, u"Bætti við athugasemd.")
a.create_item("comment_add", "", texti)
return t,a
def changelog_entry(self, notandi, skyring):
"""Examines updated fields of this instance and generates a
changelog entry,
preserving oldv values in the table atburdur_items."""
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',
'cc', 'verknumer', 'tengilidur', 'bidur_eftir'):
history = getattr(Verkefni, fld).get_history(instance, passive=True)
if history.is_modified():
old, new = None, None
if len(history.deleted_items()):
old = history.deleted_items()[0]
if len(history.added_items()):
new = 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)
for fld,key in (('fasi', 'fasi'), ('abyrgdarmadur',
'notandi'), ('mikilvaegi', 'mikilvaegi')):
history = getattr(Verkefni, fld).get_history(self, passive=True)
if history.is_modified():
old, new = None, None
if len(history.deleted_items()):
old = getattr(history.deleted_items()[0], key)
if len(history.added_items()):
new = getattr(history.added_items()[0], key)
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.parent.verkefni)
for i in history.deleted_items():
changelog('dep_first_remove', i.parent.verkefni, None)
history = Verkefni.depend_on_me.get_history(instance, passive=True)
for i in history.added_items():
changelog('dep_then_add', None, i.child.verkefni)
for i in history.deleted_items():
changelog('dep_then_remove', i.child.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)
return entry
assign_mapper(ctx, Verkefni, verkefni, properties={
'_mikilvaegi': verkefni.c.mikilvaegi,
'mikilvaegi': relation(Mikilvaegi, lazy=False),
'_fasi': verkefni.c.fasi,
'fasi': relation(Fasi, lazy=False),
'_abyrgdarmadur': verkefni.c.abyrgdarmadur,
'abyrgdarmadur': relation(Notandi,
primaryjoin=verkefni.c.abyrgdarmadur==Notandi.c.notandi, lazy=False),
'_skrad_af': verkefni.c.skrad_af,
'skrad_af': relation(Notandi,
primaryjoin=verkefni.c.skrad_af==Notandi.c.notandi, lazy=False),
'flokkar': relation(Flokkur, secondary=verkefni_flokkar, lazy=False)
})
# Atburdaskra = event log
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)
)
# Atburdur = Event
class Atburdur(base_model):
def __init__(self, verkefni, notandi, skyring=None):
self.verkefni = verkefni
self.notandi = notandi
self.skyring = skyring
self.dags = datetime.datetime.now()
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)
})
# This table stores the fields of a changelog entry.
# atburdur = event
# svid = fieldname
# gamalt = old value
# nytt = new value
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',
'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' %
(getattr_if_not_none(i.gamalt and Fasi.get(i.gamalt), "heiti"),
getattr_if_not_none(i.nytt and Fasi.get(i.nytt), "heiti"))),
'mikilvaegi': lambda i: (u'Mikilvægi: %s -> %s' %
(getattr_if_not_none(i.gamalt and Mikilvaegi.get(i.gamalt), "heiti"),
getattr_if_not_none(i.nytt and Mikilvaegi.get(i.nytt), "heiti"))),
'abyrgdarmadur': lambda i: (u'Ábyrgðarmaður: %s -> %s' %
(getattr_if_not_none(i.gamalt and Notandi.get(i.gamalt), "nafn"),
getattr_if_not_none(i.nytt and Notandi.get(i.nytt), "nafn"))),
}
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 text is not None and type(text) in (str,unicode) and
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))
})
--~--~---------~--~----~------------~-------~--~----~
You received this message because you are subscribed to the Google Groups
"sqlalchemy" group.
To post to this group, send email to sqlalchemy@googlegroups.com
To unsubscribe from this group, send email to [EMAIL PROTECTED]
For more options, visit this group at
http://groups.google.com/group/sqlalchemy?hl=en
-~----------~----~----~----~------~----~------~--~---
