9

我想在生产环境中自动生成通过 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 语句是什么？
4

4 回答 4

3

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

如何检测数据库更改

和的换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)

IPython

通过比较检查是否使用了 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 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)
于 2022-01-20T18:08:12.950 回答
1

基于aaron 的回答和内置的 IPython 魔术 %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)
于 2022-01-27T17:42:09.740 回答
1

我会考虑这样的事情：

看：

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. 您最终可能会获得比您实际需要的更多的历史记录，但至少有足够的上下文来了解正在发生的事情。

于 2022-01-19T15:22:17.843 回答
0

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

有关信号的更多文档

于 2022-01-19T09:20:17.207 回答