diff --git a/src/wutta_continuum/conf.py b/src/wutta_continuum/conf.py index 361131a..6c9abb4 100644 --- a/src/wutta_continuum/conf.py +++ b/src/wutta_continuum/conf.py @@ -80,6 +80,9 @@ class WuttaContinuumConfigExtension(WuttaConfigExtension): [wutta_continuum] wutta_plugin_spec = poser.db.continuum:PoserContinuumPlugin + + See also the SQLAlchemy-Continuum docs for + :doc:`sqlalchemy-continuum:plugins`. """ # only do this if config enables it if not config.get_bool( @@ -113,44 +116,64 @@ class WuttaContinuumPlugin(Plugin): """ SQLAlchemy-Continuum manager plugin for Wutta-Continuum. - This tries to assign the current user and IP address to the - transaction. + This is the default plugin used within + :meth:`~WuttaContinuumConfigExtension.startup()` unless config + overrides. - It will assume the "current machine" IP address, which may be - suitable for some apps but not all (e.g. web apps, where IP - address should reflect an arbitrary client machine). - - However it does not actually have a way to determine the current - user. WuttaWeb therefore uses a different plugin, based on this - one, to get both the user and IP address from current request. - - You can override this to use a custom plugin for this purpose; if - so you must specify in your config file: - - .. code-block:: ini - - [wutta_continuum] - wutta_plugin_spec = poser.db.continuum:PoserContinuumPlugin + This tries to establish the user and IP address responsible, and + comment if applicable, for the current transaction. See also the SQLAlchemy-Continuum docs for :doc:`sqlalchemy-continuum:plugins`. """ - def get_remote_addr( # pylint: disable=empty-docstring,unused-argument - self, uow, session - ): - """ """ + def get_remote_addr(self, uow, session): # pylint: disable=unused-argument + """ + This should return the effective IP address responsible for + the current change(s). + + Default logic will assume the "current machine" e.g. where a + CLI command or script is running. In practice that often + means this winds up being ``127.0.0.1`` or similar. + + :returns: IP address (v4 or v6) as string + """ host = socket.gethostname() return socket.gethostbyname(host) - def get_user_id( # pylint: disable=empty-docstring,unused-argument - self, uow, session - ): - """ """ + def get_user_id(self, uow, session): # pylint: disable=unused-argument + """ + This should return the effective ``User.uuid`` indicating who + is responsible for the current change(s). + + Default logic does not have a way to determine current user on + its own per se. However it can inspect the session, and use a + value from there if found. + + Any session can therefore declare the resonsible user:: + + myuser = session.query(model.User).first() + session.info["continuum_user_id"] = myuser.uuid + + :returns: :attr:`wuttjamaican.db.model.auth.User.uuid` value, + or ``None`` + """ + if user_id := session.info.get("continuum_user_id"): + return user_id + return None - def transaction_args(self, uow, session): # pylint: disable=empty-docstring - """ """ + def transaction_args(self, uow, session): + """ + This is a standard hook method for SQLAchemy-Continuum + plugins. We use it to (try to) inject these values, which + then become set on the current (new) transaction: + + * ``remote_addr`` - effective IP address causing the change + * see :meth:`get_remote_addr()` + * ``user_id`` - effective ``User.uuid`` for change authorship + * see :meth:`get_user_id()` + """ kwargs = {} remote_addr = self.get_remote_addr(uow, session) diff --git a/tests/test_conf.py b/tests/test_conf.py index 697592a..03485c2 100644 --- a/tests/test_conf.py +++ b/tests/test_conf.py @@ -55,9 +55,20 @@ class TestWuttaContinuumPlugin(DataTestCase): self.assertEqual(plugin.get_remote_addr(None, self.session), "127.0.0.1") def test_user_id(self): + model = self.app.model plugin = self.make_plugin() + + fred = model.User(username="fred") + self.session.add(fred) + self.session.commit() + + # empty by default self.assertIsNone(plugin.get_user_id(None, self.session)) + # but session can declare one + self.session.info["continuum_user_id"] = fred.uuid + self.assertEqual(plugin.get_user_id(None, self.session), fred.uuid) + def test_transaction_args(self): plugin = self.make_plugin() with patch.object(socket, "gethostbyname", return_value="127.0.0.1"):