wutta/wuttaweb
3
0
Fork 0
Code Issues Activity

feat: add way to declare related versions for history in MasterView

Browse source
This commit is contained in:
Lance Edgar 2026-02-27 12:59:20 -06:00
parent 205a1f7a65
commit cba8e4774d
3 changed files with 121 additions and 7 deletions
Download patch file Download diff file Expand all files Collapse all files

2
pyproject.toml
View file

@ -61,7 +61,7 @@ dependencies = [
[project.optional-dependencies]
continuum = ["Wutta-Continuum>=0.3.0"]
continuum = ["Wutta-Continuum>=0.3.3"]
docs = ["Sphinx", "furo", "sphinxcontrib-programoutput"]
tests = ["pylint", "pytest", "pytest-cov", "tox", "WebTest"]

92
src/wuttaweb/views/master.py
View file

@ -1248,8 +1248,11 @@ class MasterView(View): # pylint: disable=too-many-public-methods
Default query will locate SQLAlchemy-Continuum ``transaction``
records which are associated with versions of the given model
instance. See also
:func:`wutta-continuum:wutta_continuum.util.model_transaction_query()`.
instance. See also:
* :meth:`get_version_joins()`
* :meth:`normalize_version_joins()`
* :func:`~wutta-continuum:wutta_continuum.util.model_transaction_query()`
:returns: :class:`~sqlalchemy:sqlalchemy.orm.Query` instance
"""
@ -1260,9 +1263,67 @@ class MasterView(View): # pylint: disable=too-many-public-methods
model_class = self.get_model_class()
txncls = continuum.transaction_class(model_class)
query = model_transaction_query(instance)
query = model_transaction_query(instance, joins=self.normalize_version_joins())
return query.order_by(txncls.issued_at.desc())
def get_version_joins(self):
"""
Override this method to declare additional version tables
which should be joined when showing the overall revision
history for a given model instance.
Note that whatever this method returns, will be ran through
:meth:`normalize_version_joins()` before being passed along to
:func:`~wutta-continuum:wutta_continuum.util.model_transaction_query()`.
:returns: List of version joins info as described below.
In the simple scenario where an "extension" table is involved,
e.g. a ``UserExtension`` table::
def get_version_joins(self):
model = self.app.model
return super().get_version_joins() + [
model.UserExtension,
]
In the case where a secondary table is "related" to the main
model table, but not a standard extension (using the
``User.person`` relationship as example)::
def get_version_joins(self):
model = self.app.model
return super().get_version_joins() + [
(model.Person, "uuid", "person_uuid"),
]
See also :meth:`get_version_grid_data()`.
"""
return []
def normalize_version_joins(self):
"""
This method calls :meth:`get_version_joins()` and normalizes
the result, which will then get passed along to
:func:`~wutta-continuum:wutta_continuum.util.model_transaction_query()`.
Subclass should (generally) not override this, but instead
override :meth:`get_version_joins()`.
Each element in the return value (list) will be a 3-tuple
conforming to what is needed for the query function.
See also :meth:`get_version_grid_data()`.
:returns: List of version joins info.
"""
joins = []
for join in self.get_version_joins():
if not isinstance(join, tuple):
join = (join, "uuid", "uuid")
joins.append(join)
return joins
def configure_version_grid(self, g):
"""
Configure the grid for the :meth:`view_versions()` view.
@ -1326,7 +1387,9 @@ class MasterView(View): # pylint: disable=too-many-public-methods
model_class = self.get_model_class()
route_prefix = self.get_route_prefix()
txncls = continuum.transaction_class(model_class)
transactions = model_transaction_query(instance)
transactions = model_transaction_query(
instance, joins=self.normalize_version_joins()
)
txnid = self.request.matchdict["txnid"]
txn = transactions.filter(txncls.id == txnid).first()
@ -1408,15 +1471,34 @@ class MasterView(View): # pylint: disable=too-many-public-methods
:returns: List of version records.
"""
import sqlalchemy_continuum as continuum # pylint: disable=import-outside-toplevel
session = self.Session()
vercls = self.get_model_version_class()
return (
versions = []
# first get all versions for the model instance proper
versions.extend(
session.query(vercls)
.filter(vercls.transaction == transaction)
.filter(vercls.uuid == instance.uuid)
.all()
)
# then get all related versions, per declared joins
for child_class, foreign_attr, primary_attr in self.normalize_version_joins():
child_vercls = continuum.version_class(child_class)
versions.extend(
session.query(child_vercls)
.filter(child_vercls.transaction == transaction)
.filter(
getattr(child_vercls, foreign_attr)
== getattr(instance, primary_attr)
)
)
return versions
##############################
# autocomplete methods
##############################

34
tests/views/test_master.py
View file

@ -2212,6 +2212,26 @@ class TestVersionedMasterView(VersionWebTestCase):
view = self.make_view()
self.assertEqual(view.get_version_grid_columns(), ["issued_at", "user"])
def test_get_version_joins(self):
view = self.make_view()
self.assertEqual(view.get_version_joins(), [])
def test_normalize_version_joins(self):
model = self.app.model
view = self.make_view()
joins = [(model.Person, "uuid", "person_uuid")]
with patch.object(view, "get_version_joins", return_value=joins):
normal = view.normalize_version_joins()
self.assertEqual(normal, joins)
self.assertEqual(normal, [(model.Person, "uuid", "person_uuid")])
joins = [model.Person]
with patch.object(view, "get_version_joins", return_value=joins):
normal = view.normalize_version_joins()
self.assertNotEqual(normal, joins)
self.assertEqual(normal, [(model.Person, "uuid", "uuid")])
def test_get_version_grid_data(self):
model = self.app.model
@ -2300,7 +2320,9 @@ class TestVersionedMasterView(VersionWebTestCase):
txncls = continuum.transaction_class(model.User)
vercls = continuum.version_class(model.User)
user = model.User(username="fred")
person = model.Person(full_name="Fred Flintstone")
self.session.add(person)
user = model.User(username="fred", person=person)
self.session.add(user)
self.session.commit()
@ -2314,11 +2336,21 @@ class TestVersionedMasterView(VersionWebTestCase):
with patch.object(mod.MasterView, "model_class", new=model.User):
with patch.object(mod.MasterView, "Session", return_value=self.session):
view = self.make_view()
# just one version if no joins are specified
versions = view.get_relevant_versions(txn, user)
self.assertEqual(len(versions), 1)
version = versions[0]
self.assertIsInstance(version, vercls)
# but two versions if we specify join
joins = [(model.Person, "uuid", "person_uuid")]
with patch.object(view, "get_version_joins", return_value=joins):
versions = view.get_relevant_versions(txn, user)
self.assertEqual(len(versions), 2)
types = sorted([v.__class__.__name__ for v in versions])
self.assertEqual(types, ["PersonVersion", "UserVersion"])
def test_view_version(self):
import sqlalchemy_continuum as continuum