如何记录通过 Django shell 所做的生产数据库更改

Question

我想自动生成某种日志，记录在生产环境中通过 Django shell 所做的所有数据库更改。

我们使用架构和数据迁移脚本来更改生产数据库，并且它们是受版本控制的。因此，如果我们引入了一个错误，就很容易追溯它。但是，如果团队中的开发人员通过 Django shell 更改了数据库，从而引入了一个问题，目前我们只能希望他们记住他们所做的事情或/并且我们可以在 Python shell 历史记录中找到他们的命令。

示例。让我们假设以下代码是由团队中的开发人员通过 Python shell 执行的：

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

它会更改数据库中的团队成员资格对象。我想以某种方式记录它。

我知道有一堆Django packages related to audit logging，但我只对从 Django shell 触发的更改感兴趣，我想记录更新数据的 Python 代码。

所以我想到的问题：

• 我可以记录来自 IPython 的语句，但我怎么知道是哪一个访问了数据库？
• 我可以收听所有模型的pre_save 信号以了解数据是否更改，但我如何知道源是否来自 Python shell？我怎么知道最初的 Python 语句是什么？

Answer 1

如果进行了任何数据库更改，此解决方案会记录会话中的所有命令。

如何检测数据库更改

包装execute_sql 的SQLInsertCompiler、SQLUpdateCompiler 和SQLDeleteCompiler。

SQLDeleteCompiler.execute_sql 返回一个游标包装器。

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)

如何记录通过 Django shell 发出的命令

atexit.register() 一个执行 readline.write_history_file() 的退出处理程序。

import atexit
import readline

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

atexit.register(exit_handler)

IPython

通过比较HistoryAccessor.get_last_session_id()来判断是否使用了IPython。

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)

把它们放在一起

在manage.py中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

基于the answer of aaron和the built-in IPython magic %logstart的实现，这就是我们最终想出的方案。

如果任何命令通过 Django ORM 触发了数据库写入，则最后一次 IPython 会话的所有命令都会记录在历史文件中。

这是生成的历史文件的摘录：

❯ 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]# 'test-albero-jose-maria@gmail.com'
# Thu, 27 Jan 2022 16:20:28
p.save()

现在是我们的manage.py：

#!/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)

Answer 3

我会考虑这样的事情：

• 使用某种初始化代码来包装每个 python 会话，例如 PYTHONSTARTUP 环境变量 https://docs.python.org/3/using/cmdline.html#envvar-PYTHONSTARTUP

• 在PYTHONSTARTUP 指向使用atexit 注册退出处理程序的文件中 https://docs.python.org/3/library/atexit.html

• 这两件事应该允许您使用一些较低级别的 API django-reversion 包装整个终端会话 https://django-reversion.readthedocs.io/en/stable/api.html#creating-revisions（类似这样，但在您的启动和atexit 代码中直接调用该上下文管理器的__enter__ 和__exit__）。不幸的是我不知道细节，但它应该是可行的。

• 在atexit/revision end 调用代码列出附加行 终端会话并将它们存储在数据库中的其他位置，并参考特定修订。

见：

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

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

基本上，您可以调用get_history_length 两次：在终端会话的开始和结束时。这将允许您使用get_history_item 获得更改发生位置的相关行。您最终可能会获得比您实际需要的更多的历史记录，但至少有足够的上下文来了解正在发生的事情。

Answer 4

您可以使用 django 的 receiver 注释。

例如，如果你想检测任何对save方法的调用，你可以这样做：

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")

更多documentation for the signals