Welcome | Get started | Dive | Contribute | Topics | Reference | Changes | More
watch – Watching database changes¶
The lino_book.projects.watch demo project illustrates some local
customizations.
This tutorial explains how to use the lino.modlib.changes plugin for
logging changes to individual rows of database tables and implementing a kind of
audit trail.
This tutorial is a tested document and
uses the lino_book.projects.watch sample application:
>>> from lino import startup
>>> startup('lino_book.projects.watch.settings')
To enable database change watching, you add lino.modlib.changes
to your get_installed_plugins and then register “change
watchers” for every type of change you want to watch.
The example in this tutorial uses the lino_xl.lib.contacts
module. It also adds a model Entry as an example of a watched slave model.
The “master” of a change watcher is the object to which every change should be attributed. In this example the master is Partner: every tracked change to Entry, Partner or Company will be assigned to their Partner record.
In the settings.py file we define our own subclass of Site, with a setup_actions method to call watch_changes.
from lino.projects.std.settings import *
class Site(Site):
is_demo_site = True
demo_fixtures = "std demo demo2"
languages = 'en'
# default_user = "robin"
user_types_module = 'lino_xl.lib.xl.user_types'
# default_ui = "lino_react.react"
def get_installed_plugins(self):
yield super().get_installed_plugins()
yield 'lino_xl.lib.contacts'
yield 'lino.modlib.changes'
yield 'lino.modlib.users'
yield 'lino_book.projects.watch.entries'
# yield 'lino_xl.lib.notes'
def setup_actions(self):
# watch changes to Partner, Company and Entry
# objects, grouped to their respective Partner.
super().setup_actions()
from lino.modlib.changes.utils import watch_changes as wc
wc(self.models.contacts.Partner)
wc(self.models.contacts.Company, master_key='partner_ptr')
wc(self.models.entries.Entry, master_key='company__partner_ptr')
# wc(self.models.notes.Note, master_key='company__partner_ptr')
SITE = Site(globals())
DEBUG = True
The entries/models.py file defines the Entry model and its data tables:
from django.db import models
from lino.api import dd, _
from lino.modlib.users.mixins import My, UserAuthored
class Entry(UserAuthored):
class Meta:
verbose_name = _("Entry")
verbose_name_plural = _("Entries")
subject = models.CharField(_("Subject"), blank=True, max_length=200)
body = dd.RichTextField(_("Body"), blank=True)
company = dd.ForeignKey('contacts.Company')
class Entries(dd.Table):
model = Entry
detail_layout = """
id user
company
subject
body
"""
insert_layout = """
company
subject
"""
class EntriesByCompany(Entries):
master_key = 'company'
class MyEntries(My, Entries):
pass
# @dd.receiver(dd.post_startup)
# def my_change_watchers(sender, **kw):
The tests/test_basics.py file contains a unit test:
# -*- coding: utf-8 -*-
# Copyright 2017-2025 Rumma & Ko Ltd
# License: GNU Affero General Public License v3 (see file COPYING for details)
# $ pm test tests.test_basics
"""Basic tests.
"""
from django.conf import settings
from lino.utils.djangotest import RemoteAuthTestCase
from lino.core.callbacks import popCallBack, applyCallbackChoice
class QuickTest(RemoteAuthTestCase):
# fixtures = ['demo', 'demo2']
maxDiff = None
def test_this(self):
from lino.api import rt
from lino.core.renderer import TestRenderer
UserTypes = rt.models.users.UserTypes
rt.models.users.User(username="robin",
user_type=UserTypes.admin).save()
ses = rt.login('robin',
renderer=TestRenderer(
settings.SITE.kernel.web_front_ends[0]))
s = ses.show('changes.Changes')
self.assertEqual(s, "No data to display")
rr = rt.models.contacts.Companies.required_roles
self.assertTrue(ses.user.user_type.role.satisfies_requirement(rr))
# We create a new organization:
url = '/api/contacts/Companies'
data = dict(an='submit_insert', name='My pub')
res = self.post_json_dict('robin', url, data)
self.assertEqual(res.message,
'Organization "My pub" has been created.')
s = ses.show('changes.Changes',
column_names="id type master object diff")
# print(s)
expected = """\
==== ============= ================ ================ ===========================================
ID Change Type Master Object Changes
---- ------------- ---------------- ---------------- -------------------------------------------
1 Create `My pub <…>`__ `My pub <…>`__ Company(id=1,name='My pub',partner_ptr=1)
==== ============= ================ ================ ===========================================
"""
self.assertEqual(s, expected)
url = '/api/contacts/Companies/1'
data = "an=submit_detail&name=Our%20pub"
res = self.put_json_dict('robin', url, data)
self.assertEqual(res.message,
'Organization "Our pub" has been updated.')
output = ses.show('changes.Changes',
column_names="id type master object diff")
# print(output)
expected = """\
==== ============= ================= ================= ===========================================
ID Change Type Master Object Changes
---- ------------- ----------------- ----------------- -------------------------------------------
2 Update `Our pub <…>`__ `Our pub <…>`__ name : 'My pub' --> 'Our pub'
1 Create `Our pub <…>`__ `Our pub <…>`__ Company(id=1,name='My pub',partner_ptr=1)
==== ============= ================= ================= ===========================================
"""
self.assertEqual(output, expected)
# We add an entry:
url = '/api/entries/Entries'
data = dict(an='submit_insert', subject='test', companyHidden=1)
res = self.post_json_dict('robin', url, data)
self.assertEqual(res.message,
'Entry "Entry object (1)" has been created.')
output = ses.show('changes.Changes',
column_names="id type master object diff")
# print(output)
expected = """\
==== ============= ================= ========================== =============================================
ID Change Type Master Object Changes
---- ------------- ----------------- -------------------------- ---------------------------------------------
3 Create `Our pub <…>`__ `Entry object (1) <…>`__ Entry(id=1,user=1,subject='test',company=1)
2 Update `Our pub <…>`__ `Our pub <…>`__ name : 'My pub' --> 'Our pub'
1 Create `Our pub <…>`__ `Our pub <…>`__ Company(id=1,name='My pub',partner_ptr=1)
==== ============= ================= ========================== =============================================
"""
self.assertEqual(output, expected)
# Now we delete the entry:
url = '/api/entries/Entries/1'
data = dict(an='delete_selected', sr=1)
res = self.get_json_dict('robin', url, data)
self.assertEqual(
res.message, """\
You are about to delete 1 Entry
(Entry object (1)). Are you sure?""")
# We answer "yes":
applyCallbackChoice(res, data, "yes")
# url = "/callbacks/{0}/yes".format(res['xcallback']['id'])
# print(data)
res = self.get_json_dict('robin', url, data)
# print(res)
# r = test_client.get(url)
self.assertEqual(res.success, True)
self.assertEqual(res.record_deleted, True)
output = ses.show('changes.Changes',
column_names="id type master object diff")
# print(output)
expected = """\
==== ============= ================= ================= =============================================
ID Change Type Master Object Changes
---- ------------- ----------------- ----------------- ---------------------------------------------
4 Delete `Our pub <…>`__ Entry(id=1,user=1,subject='test',company=1)
3 Create `Our pub <…>`__ Entry(id=1,user=1,subject='test',company=1)
2 Update `Our pub <…>`__ `Our pub <…>`__ name : 'My pub' --> 'Our pub'
1 Create `Our pub <…>`__ `Our pub <…>`__ Company(id=1,name='My pub',partner_ptr=1)
==== ============= ================= ================= =============================================
"""
self.assertEqual(output, expected)
# Note how the `object` column of the first two rows in above
# table is empty. That's because the entry object has been
# deleted, so it does no longer exist in the database and Lino
# cannot point to it. But note also that `object` is a
# "nullable Generic ForeignKey", the underlying fields
# `object_id` and `object_type` still contain their
# values. Here is the same table with "Object" split into its
# components:
output = ses.show(
'changes.Changes',
column_names="id type master object_type object_id diff")
# print(output)
expected = """\
==== ============= ================= ============== =========== =============================================
ID Change Type Master Object type object id Changes
---- ------------- ----------------- -------------- ----------- ---------------------------------------------
4 Delete `Our pub <…>`__ Entry 1 Entry(id=1,user=1,subject='test',company=1)
3 Create `Our pub <…>`__ Entry 1 Entry(id=1,user=1,subject='test',company=1)
2 Update `Our pub <…>`__ Organization 1 name : 'My pub' --> 'Our pub'
1 Create `Our pub <…>`__ Organization 1 Company(id=1,name='My pub',partner_ptr=1)
==== ============= ================= ============== =========== =============================================
"""
self.assertEqual(output, expected)
# Until 20150626 only the
# :attr:`object<lino.modlib.changes.models.Change.object>` was nullable,
# not the :attr:`master<lino.modlib.changes.models.Change.master>`. But
# now you can also delete the master, and all change records will still
# remain:
url = '/api/contacts/Companies/1'
data = dict(an='delete_selected', sr=1)
res = self.get_json_dict('robin', url, data)
applyCallbackChoice(res, data, "yes")
# url = "/callbacks/{0}/yes".format(res.xcallback['id'])
self.get_json_dict('robin', url, data)
output = ses.show('changes.Changes',
column_names="id type master object diff")
# print(output)
expected = """\
==== ============= ======== ======== =============================================
ID Change Type Master Object Changes
---- ------------- -------- -------- ---------------------------------------------
5 Delete Company(id=1,name='Our pub',partner_ptr=1)
4 Delete Entry(id=1,user=1,subject='test',company=1)
3 Create Entry(id=1,user=1,subject='test',company=1)
2 Update name : 'My pub' --> 'Our pub'
1 Create Company(id=1,name='My pub',partner_ptr=1)
==== ============= ======== ======== =============================================
"""
self.assertEqual(output, expected)
# Of course these change records are now considered broken GFKs:
output = ses.show('gfks.BrokenGFKs')
# print(output)
expected = """\
================ ================= =========================================================== ========
Database model Database object Message Action
---------------- ----------------- ----------------------------------------------------------- --------
`Change <…>`__ `#1 <…>`__ Invalid primary key 1 for contacts.Partner in `master_id` clear
`Change <…>`__ `#2 <…>`__ Invalid primary key 1 for contacts.Partner in `master_id` clear
`Change <…>`__ `#3 <…>`__ Invalid primary key 1 for contacts.Partner in `master_id` clear
`Change <…>`__ `#4 <…>`__ Invalid primary key 1 for contacts.Partner in `master_id` clear
`Change <…>`__ `#5 <…>`__ Invalid primary key 1 for contacts.Partner in `master_id` clear
`Change <…>`__ `#1 <…>`__ Invalid primary key 1 for contacts.Company in `object_id` clear
`Change <…>`__ `#2 <…>`__ Invalid primary key 1 for contacts.Company in `object_id` clear
`Change <…>`__ `#3 <…>`__ Invalid primary key 1 for entries.Entry in `object_id` clear
`Change <…>`__ `#4 <…>`__ Invalid primary key 1 for entries.Entry in `object_id` clear
`Change <…>`__ `#5 <…>`__ Invalid primary key 1 for contacts.Company in `object_id` clear
================ ================= =========================================================== ========
"""
self.assertEqual(output, expected)
# There open questions regarding these change records:
# - Do we really never want to remove them? Do we really want a nullable
# master field? Should this option be configurable?
# - How to tell :class:`lino.modlib.gfks.models.BrokenGFKs` to
# differentiate them from ?
# - Should :meth:`get_broken_generic_related
# <lino.core.kernel.Kernel.get_broken_generic_related>` suggest to
# "clear" nullable GFK fields?
Here is our demo fixture, which partly reproduces what we are doing in the temporary database during djangotests:
from lino.api import rt
from lino.core.diff import ChangeWatcher
from lino.core.utils import PseudoRequest
def objects():
Company = rt.models.contacts.Company
ar = rt.login(request=PseudoRequest("robin"))
obj = Company(name="My pub")
obj.full_clean()
obj.save_new_instance(ar)
cw = ChangeWatcher(obj)
obj.name = "Our pub"
obj.save_watched_instance(ar, cw)
# obj.delete_instance(ar)
# this is a special fixture : it creates objects as a side effect
# but does not yield them.
return []
>>> from lino.api.doctest import *
>>> rt.show('changes.Changes')
...
==== ============ ============================ ============= ============== =========== ============= =========== ============================================= ======== ================= =================
ID Author time Change Type Object type object id Master type master id Changes Fields Object Master
---- ------------ ---------------------------- ------------- -------------- ----------- ------------- ----------- --------------------------------------------- -------- ----------------- -----------------
2 Robin Rood ... Update Organization 82 Partner 82 name : 'My pub' --> 'Our pub' name `Our pub <…>`__ `Our pub <…>`__
1 Robin Rood ... Create Organization 82 Partner 82 Company(id=82,name='My pub',partner_ptr=82) `Our pub <…>`__ `Our pub <…>`__
==== ============ ============================ ============= ============== =========== ============= =========== ============================================= ======== ================= =================