Add user defined callback to be called on every shell execution

README.rst

@@ -232,6 +232,35 @@ If you want to persist indefinitly all declared variables on the shell, set this
 
 **BEWARE**: *leaving this disabled is not recomended on production code!*
 
+ADMIN_SHELL_CALLBACK
+^^^^^^^^^^^^^^^^^^^^^^^^^^^
+*type* : **string**
+
+*default* : **None**
+
+This setting allows you to specify a callback function that will be called after each code execution in the admin shell. The callback function should be provided as a string in the format "path.to.module.function_name".
+
+The callback function will receive a dictionary with the following keys:
+- 'request': The current request object
+- 'user': The user who executed the code
+- 'code': The code that was executed
+- 'response': The result of the code execution
+- 'timestamp': The time when the code was executed
+
+This can be useful for logging or auditing purposes (e.g. saving executed code and its result to database). For example:
+
+.. code-block:: python
+
+    def my_callback(callback_data):
+        print(f"User {callback_data['user']} executed code at {callback_data['timestamp']}")
+
+    # In your settings.py
+    ADMIN_SHELL_CALLBACK = 'path.to.my_callback'
+
+If the callback function is defined but cannot be imported or raises an exception, a warning will be logged.
+
+**Note**: Be careful when implementing the callback function, as it will be called for every code execution in the admin shell.
+
 
 Code examples
 -------------

django_admin_shell/settings.py

@@ -72,3 +72,8 @@ def from_settings_or_default(name, default):
     'ADMIN_SHELL_CLEAR_SCOPE_ON_CLEAR_HISTORY',
     False
 )
+
+ADMIN_SHELL_CALLBACK = from_settings_or_default(
+    'ADMIN_SHELL_CALLBACK',
+    None
+)

django_admin_shell/tests/test_view.py

@@ -208,3 +208,62 @@ def test_clear_scope(self):
         assert ADMIN_SHELL_SESSION_KEY in self.client_auth.session
         assert self.client_auth.session[ADMIN_SHELL_SESSION_KEY] == []
         assert 'a' not in self.view.runner.importer.get_scope()
+
+    @override_settings(DEBUG=True)
+    @mock.patch(
+        'django_admin_shell.views.ADMIN_SHELL_CALLBACK',
+        'django_admin_shell.tests.utils.callback')
+    @mock.patch('django_admin_shell.tests.utils.callback')
+    def test_callback_function(self, mock_callback) -> None:
+        """
+        Show that the callback function is called with the correct arguments.
+        """
+        self.user.is_staff = True
+        self.user.is_superuser = True
+        self.user.save()
+
+        response = self.client_auth.post(self.url, {"code": 'print("Hello, World!")'})
+
+        assert mock_callback.called
+
+        callback_data = mock_callback.call_args[0][0]
+        assert 'request' in callback_data
+        assert 'user' in callback_data
+        assert 'code' in callback_data
+        assert 'response' in callback_data
+        assert 'timestamp' in callback_data
+        assert callback_data['code'] == 'print("Hello, World!")'
+        assert callback_data['response']['status'] == 'success'
+        assert callback_data['response']['out'] == 'Hello, World!\n'
+
+        assert response.status_code == 302
+        session = self.client_auth.session[ADMIN_SHELL_SESSION_KEY]
+        assert len(session) == 1
+        assert session[0]['code'] == 'print("Hello, World!")'
+        assert session[0]['status'] == 'success'
+        assert session[0]['out'] == 'Hello, World!\n'
+
+    @override_settings(DEBUG=True)
+    @mock.patch(
+        'django_admin_shell.views.ADMIN_SHELL_CALLBACK',
+        'django_admin_shell.tests.utils.callback')
+    @mock.patch('django_admin_shell.tests.utils.callback')
+    def test_callback_function_error(self, mock_callback) -> None:
+        self.user.is_staff = True
+        self.user.is_superuser = True
+        self.user.save()
+
+        def mock_callback_error(callback_data):
+            raise Exception("Test exception")
+
+        mock_callback.side_effect = mock_callback_error
+
+        with self.assertWarns(RuntimeWarning):
+            response = self.client_auth.post(self.url, {"code": 'print("Hello, World!")'})
+
+        assert response.status_code == 302
+        session = self.client_auth.session[ADMIN_SHELL_SESSION_KEY]
+        assert len(session) == 1
+        assert session[0]['code'] == 'print("Hello, World!")'
+        assert session[0]['status'] == 'success'
+        assert session[0]['out'] == 'Hello, World!\n'

django_admin_shell/tests/utils.py

@@ -0,0 +1,3 @@
+# Just for testing
+def callback(callback_data):
+    pass

django_admin_shell/views.py

@@ -1,12 +1,16 @@
 # encoding: utf-8
 from django.apps import apps
 from django.views.generic import FormView
+
+from django.utils.module_loading import import_string
+
 from .forms import ShellForm
 from django.http import (
     HttpResponseForbidden,
     HttpResponseNotFound
 )
 from django.conf import settings
+from django.utils import timezone
 
 try:
     # Only for python 2
@@ -24,7 +28,8 @@
     ADMIN_SHELL_IMPORT_DJANGO,
     ADMIN_SHELL_IMPORT_DJANGO_MODULES,
     ADMIN_SHELL_IMPORT_MODELS,
-    ADMIN_SHELL_CLEAR_SCOPE_ON_CLEAR_HISTORY
+    ADMIN_SHELL_CLEAR_SCOPE_ON_CLEAR_HISTORY,
+    ADMIN_SHELL_CALLBACK
 )
 
 import django
@@ -234,11 +239,14 @@ def get(self, request, *args, **kwargs):
         return super(ShellView, self).get(request, *args, **kwargs)
 
     def form_valid(self, form):
+
         code = form.cleaned_data.get("code", "")
+        result = None
         if len(code.strip()) > 0:
             result = self.runner.run_code(code)
             self.add_to_outout(result)
             self.save_output()
+        self.call_callback(self.request, result, code)
         return super(ShellView, self).form_valid(form)
 
     def get_context_data(self, **kwargs):
@@ -251,3 +259,36 @@ def get_context_data(self, **kwargs):
         ctx['django_version'] = get_dj_version()
         ctx['auto_import'] = str(self.runner.importer)
         return ctx
+
+    def call_callback(self, request, response, code) -> None:
+        callback_string = ADMIN_SHELL_CALLBACK
+        if not callback_string:
+            return
+        try:
+            callback = import_string(callback_string)
+        except Exception as e:
+            warnings.warn(
+                f"Error in trying to import callback function: {str(e)}",
+                RuntimeWarning
+            )
+            return
+        if not callable(callback):
+            warnings.warn(
+                f"ADMIN_SHELL_CALLBACK is set but is not callable: {callback_string}",
+                RuntimeWarning
+            )
+            return
+        try:
+            callback_data = {
+                'request': request,
+                'user': request.user,
+                'code': code,
+                'response': response,
+                'timestamp': timezone.now()
+            }
+            callback(callback_data)
+        except Exception as e:
+            warnings.warn(
+                f"Error in ADMIN_SHELL_CALLBACK: {str(e)}",
+                RuntimeWarning
+            )