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-2019 Rumma & Ko Ltd
# License: GNU Affero General Public License v3 (see file COPYING for details)
"""Basic tests.
This module is part of the Lino test suite. You can test only this
module by issuing either::
$ go watch
$ python manage.py test tests.test_basics
"""
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=100,name='My pub',partner_ptr=100)
==== ============= ================ ================ ===============================================
"""
self.assertEqual(s, expected)
url = '/api/contacts/Companies/100'
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=100,name='My pub',partner_ptr=100)
==== ============= ================= ================= ===============================================
"""
self.assertEqual(output, expected)
# We add an entry:
url = '/api/entries/Entries'
data = dict(an='submit_insert', subject='test', companyHidden=100)
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=100)
2 Update `Our pub <…>`__ `Our pub <…>`__ name : 'My pub' --> 'Our pub'
1 Create `Our pub <…>`__ `Our pub <…>`__ Company(id=100,name='My pub',partner_ptr=100)
==== ============= ================= ========================== ===============================================
"""
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=100)
3 Create `Our pub <…>`__ Entry(id=1,user=1,subject='test',company=100)
2 Update `Our pub <…>`__ `Our pub <…>`__ name : 'My pub' --> 'Our pub'
1 Create `Our pub <…>`__ `Our pub <…>`__ Company(id=100,name='My pub',partner_ptr=100)
==== ============= ================= ================= ===============================================
"""
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=100)
3 Create `Our pub <…>`__ Entry 1 Entry(id=1,user=1,subject='test',company=100)
2 Update `Our pub <…>`__ Organization 100 name : 'My pub' --> 'Our pub'
1 Create `Our pub <…>`__ Organization 100 Company(id=100,name='My pub',partner_ptr=100)
==== ============= ================= ============== =========== ===============================================
"""
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/100'
data = dict(an='delete_selected', sr=100)
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=100,name='Our pub',partner_ptr=100)
4 Delete Entry(id=1,user=1,subject='test',company=100)
3 Create Entry(id=1,user=1,subject='test',company=100)
2 Update name : 'My pub' --> 'Our pub'
1 Create Company(id=100,name='My pub',partner_ptr=100)
==== ============= ======== ======== ================================================
"""
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 100 for contacts.Partner in `master_id` clear
`Change <…>`__ `#2 <…>`__ Invalid primary key 100 for contacts.Partner in `master_id` clear
`Change <…>`__ `#3 <…>`__ Invalid primary key 100 for contacts.Partner in `master_id` clear
`Change <…>`__ `#4 <…>`__ Invalid primary key 100 for contacts.Partner in `master_id` clear
`Change <…>`__ `#5 <…>`__ Invalid primary key 100 for contacts.Partner in `master_id` clear
`Change <…>`__ `#1 <…>`__ Invalid primary key 100 for contacts.Company in `object_id` clear
`Change <…>`__ `#2 <…>`__ Invalid primary key 100 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 100 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 181 Partner 181 name : 'My pub' --> 'Our pub' name `Our pub <…>`__ `Our pub <…>`__
1 Robin Rood ... Create Organization 181 Partner 181 Company(id=181,name='My pub',partner_ptr=181) `Our pub <…>`__ `Our pub <…>`__
==== ============ ============================ ============= ============== =========== ============= =========== =============================================== ======== ================= =================