如何记录通过 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 包，但我只对 Django shell 触发的更改感兴趣，并且我想记录更新数据的 Python 代码。

所以我想到的问题是：

• 我可以从 IPython 记录语句，但我如何知道哪一个接触了数据库？
• 我可以监听pre_save所有模型的信号来了解数据是否发生变化，但我如何知道源是否来自 Python shell？我怎么知道原始的 Python 语句是什么？

Answer 1

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

如何检测数据库变化

包裹,和.execute_sql​SQLInsertCompilerSQLUpdateCompilerSQLDeleteCompiler

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)

Python

通过比较检查是否使用了IPython 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)

把它们放在一起

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