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

Question

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

我们使用架构和数据迁移脚本来更改生产数据库，并且它们是受版本控制的。因此，如果我们引入了一个错误，就很容易对其进行追踪。但是，如果团队中的开发人员通过 Django shell 更改数据库，然后引入问题，目前我们只能希望他们记住他们所做的事情 or/and 我们可以在 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

您可以使用 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

Answer 2

我会考虑这样的事情：

用某种初始化代码包装每个 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 3

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

如何检测数据库变化

将 SQLInsertCompiler、SQLUpdateCompiler 和 SQLDeleteCompiler 的 execute_sql 换行。

SQLDeleteCompiler.execute_sql returns 游标包装器。

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)

把它们放在一起

在execute_from_command_line(sys.argv)前的manage.py中添加以下内容。

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 4

基于 and the implementation of 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)

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

How to log production database changes made via the Django shell

python

django

audit-logging

django-shell

如何检测数据库变化

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

IPython

把它们放在一起