How to log production database changes made via the Django shell

Question

I would like to automatically generate some sort of log of all the database changes that are made via the Django shell in the production environment.

We use schema and data migration scripts to alter the production database and they are version controlled. Therefore if we introduce a bug, it's easy to track it back. But if a developer in the team changes the database via the Django shell which then introduces an issue, at the moment we can only hope that they remember what they did or/and we can find their commands in the Python shell history.

Example. Let's imagine that the following code was executed by a developer in the team via the Python shell:

>>> tm = TeamMembership.objects.get(person=alice)
>>> tm.end_date = date(2022,1,1)
>>> tm.save()

It changes a team membership object in the database. I would like to log this somehow.

I'm aware that there are a bunch of Django packages related to audit logging, but I'm only interested in the changes that are triggered from the Django shell, and I want to log the Python code that updated the data.

So the questions I have in mind:

• I can log the statements from IPython but how do I know which one touched the database?
• I can listen to the pre_save signal for all model to know if data changes, but how do I know if the source was from the Python shell? How do I know what was the original Python statement?

Answer 1

This solution logs all commands in the session if any database changes were made.

How to detect database changes

Wrap execute_sql of SQLInsertCompiler, SQLUpdateCompiler and SQLDeleteCompiler.

SQLDeleteCompiler.execute_sql returns a cursor wrapper.

from django.db.models.sql.compiler import SQLInsertCompiler, SQLUpdateCompiler, SQLDeleteCompiler

changed = False

def check_changed(func):
    def _func(*args, **kwargs):
        nonlocal changed
        result = func(*args, **kwargs)
        if not changed and result:
            changed = not hasattr(result, 'cursor') or bool(result.cursor.rowcount)
        return result
    return _func

SQLInsertCompiler.execute_sql = check_changed(SQLInsertCompiler.execute_sql)
SQLUpdateCompiler.execute_sql = check_changed(SQLUpdateCompiler.execute_sql)
SQLDeleteCompiler.execute_sql = check_changed(SQLDeleteCompiler.execute_sql)

How to log commands made via the Django shell

atexit.register() an exit handler that does readline.write_history_file().

import atexit
import readline

def exit_handler():
    filename = 'history.py'
    readline.write_history_file(filename)

atexit.register(exit_handler)

IPython

Check whether IPython was used by comparing HistoryAccessor.get_last_session_id().

import atexit
import io
import readline

ipython_last_session_id = None
try:
    from IPython.core.history import HistoryAccessor
except ImportError:
    pass
else:
    ha = HistoryAccessor()
    ipython_last_session_id = ha.get_last_session_id()

def exit_handler():
    filename = 'history.py'
    if ipython_last_session_id and ipython_last_session_id != ha.get_last_session_id():
        cmds = '\n'.join(cmd for _, _, cmd in ha.get_range(ha.get_last_session_id()))
        with io.open(filename, 'a', encoding='utf-8') as f:
            f.write(cmds)
            f.write('\n')
    else:
        readline.write_history_file(filename)

atexit.register(exit_handler)

Put it all together

Add the following in manage.py before execute_from_command_line(sys.argv).

if sys.argv[1] == 'shell':
    import atexit
    import io
    import readline

    from django.db.models.sql.compiler import SQLInsertCompiler, SQLUpdateCompiler, SQLDeleteCompiler

    changed = False

    def check_changed(func):
        def _func(*args, **kwargs):
            nonlocal changed
            result = func(*args, **kwargs)
            if not changed and result:
                changed = not hasattr(result, 'cursor') or bool(result.cursor.rowcount)
            return result
        return _func

    SQLInsertCompiler.execute_sql = check_changed(SQLInsertCompiler.execute_sql)
    SQLUpdateCompiler.execute_sql = check_changed(SQLUpdateCompiler.execute_sql)
    SQLDeleteCompiler.execute_sql = check_changed(SQLDeleteCompiler.execute_sql)

    ipython_last_session_id = None
    try:
        from IPython.core.history import HistoryAccessor
    except ImportError:
        pass
    else:
        ha = HistoryAccessor()
        ipython_last_session_id = ha.get_last_session_id()

    def exit_handler():
        if changed:
            filename = 'history.py'
            if ipython_last_session_id and ipython_last_session_id != ha.get_last_session_id():
                cmds = '\n'.join(cmd for _, _, cmd in ha.get_range(ha.get_last_session_id()))
                with io.open(filename, 'a', encoding='utf-8') as f:
                    f.write(cmds)
                    f.write('\n')
            else:
                readline.write_history_file(filename)

    atexit.register(exit_handler)

Answer 2

You could use django's receiver annotation.

For example, if you want to detect any call of the save method, you could do:

from django.db.models.signals import post_save
from django.dispatch import receiver
import logging

@receiver(post_save)
def logg_save(sender, instance, **kwargs):
    logging.debug("whatever you want to log")

some more documentation for the signals

Answer 3

I would consider something like this:

• Wrapping each python session with some sort initialisation code using e.g. PYTHONSTARTUP environment variable https://docs.python.org/3/using/cmdline.html#envvar-PYTHONSTARTUP

• In the file where PYTHONSTARTUP points to registering Exit handler using atexit https://docs.python.org/3/library/atexit.html

• These two things should allow you to use some lower level APIs of django-reversion to wrap the whole terminal session with https://django-reversion.readthedocs.io/en/stable/api.html#creating-revisions (something like this but calling __enter__ and __exit__ of that context manager directly in your startup and atexit code). Unfortunately I don't know the details but it should be doable.

• In atexit / revision end calling the code to list the additional lines of the terminal session and storing them somewhere else in the database with a reference to the specific revision.

See:

https://docs.python.org/3/library/readline.html#readline.get_history_length

https://docs.python.org/3/library/readline.html#readline.get_history_item

Basically, the idea is that you could call get_history_length twice: at the beginning and end of the terminal session. That will allow you to get relevant lines of where the change took place using get_history_item. You may end up having more lines of history than what you actually need but at least there is enough context to see what's going on.

Answer 4

Based on the answer of aaron and the implementation of the built-in IPython magic %logstart, this is the solution we came up with in the end.

All commands of the last IPython session are logged in a history file if any of the commands triggered a database write through the Django ORM.

Here's an excerpt of the generated history file:

❯ cat ~/.python_shell_write_history
# Thu, 27 Jan 2022 16:20:28
#
# New Django shell session started
#
# Thu, 27 Jan 2022 16:20:28
from people.models import *
# Thu, 27 Jan 2022 16:20:28
p = Person.objects.first()
# Thu, 27 Jan 2022 16:20:28
p
#[Out]# <Person: Test Albero Jose Maria>
# Thu, 27 Jan 2022 16:20:28
p.email
#[Out]# '[email protected]'
# Thu, 27 Jan 2022 16:20:28
p.save()

Here's our manage.py now:

#!/usr/bin/env python
import os
import sys


def shell_audit(logfname: str) -> None:
    """If any of the Python shell commands changed the Django database during the
    session, capture all commands in a logfile for future analysis."""
    import atexit

    from django.db.models.sql.compiler import (
        SQLDeleteCompiler,
        SQLInsertCompiler,
        SQLUpdateCompiler,
    )

    changed = False

    def check_changed(func):
        def _func(*args, **kwargs):
            nonlocal changed
            result = func(*args, **kwargs)
            if not changed and result:
                changed = not hasattr(result, "cursor") or bool(result.cursor.rowcount)
            return result

        return _func

    SQLInsertCompiler.execute_sql = check_changed(SQLInsertCompiler.execute_sql)
    SQLUpdateCompiler.execute_sql = check_changed(SQLUpdateCompiler.execute_sql)
    SQLDeleteCompiler.execute_sql = check_changed(SQLDeleteCompiler.execute_sql)

    def exit_handler():
        if not changed:
            return None

        from IPython.core import getipython

        shell = getipython.get_ipython()
        if not shell:
            return None

        logger = shell.logger

        # Logic borrowed from %logstart (IPython.core.magics.logging)
        loghead = ""
        log_session_head = "#\n# New Django shell session started\n#\n"
        logmode = "append"
        log_output = True
        timestamp = True
        log_raw_input = False
        logger.logstart(logfname, loghead, logmode, log_output, timestamp, log_raw_input)

        log_write = logger.log_write
        input_hist = shell.history_manager.input_hist_parsed
        output_hist = shell.history_manager.output_hist_reprs

        log_write(log_session_head)
        for n in range(1, len(input_hist)):
            log_write(input_hist[n].rstrip() + "\n")
            if n in output_hist:
                log_write(output_hist[n], kind="output")

    atexit.register(exit_handler)


if __name__ == "__main__":
    os.environ.setdefault("DJANGO_SETTINGS_MODULE", "project.settings")
    try:
        from django.core.management import execute_from_command_line
    except ImportError:
        # The above import may fail for some other reason. Ensure that the
        # issue is really that Django is missing to avoid masking other
        # exceptions on Python 2.
        try:
            import django  # noqa: F401
        except ImportError:
            raise ImportError(
                "Couldn't import Django. Are you sure it's installed and "
                "available on your PYTHONPATH environment variable? Did you "
                "forget to activate a virtual environment?"
            )
        raise
    if sys.argv[1] == "shell":
        logfname = os.path.expanduser("~/.python_shell_write_history")
        shell_audit(logfname)
    execute_from_command_line(sys.argv)