Как регистрировать изменения в производственной базе данных, сделанные с помощью оболочки Django

Я хотел бы автоматически генерировать некий журнал всех изменений в базе данных, которые делаются через оболочку Django в производственной среде. Наша база кода контролируется по версиям, поэтому если мы вносим ошибку, ее легко отследить. Но если разработчик в команде изменяет базу данных через оболочку Django, что приводит к возникновению проблемы, в настоящее время мы можем только надеяться, что они помнят, что они сделали, или/и мы можем найти их команды в оболочке Python.

Я знаю, что существует куча пакетов Django, связанных с логированием аудита, но меня интересуют только те изменения, которые запускаются из оболочки Django. Также, в идеале, мы могли бы регистрировать и Python код, который обновил данные.

Вы можете использовать аннотацию 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")

немного больше документации для сигналов

Я бы рассмотрел что-то вроде этого:

  • Обертывание каждой сессии python каким-либо кодом инициализации, используя, например. PYTHONSTARTUP переменная окружения https://docs.python.org/3/using/cmdline.html#envvar-PYTHONSTARTUP

  • В файле, где PYTHONSTARTUP указывает на регистрацию обработчика Exit с помощью atexit https://docs.python.org/3/library/atexit.html

  • Эти две вещи должны позволить вам использовать некоторые API более низкого уровня от django-reversion, чтобы обернуть весь терминальный сеанс с https://django-reversion.readthedocs.io/en/stable/api.html#creating-revisions (что-то вроде этого, но вызывая __enter__ и __exit__ этого контекстного менеджера непосредственно в вашем коде запуска и atexit). К сожалению, я не знаю деталей, но это должно быть выполнимо.

  • В atexit / конце ревизии вызов кода для перечисления дополнительных строк терминальной сессии и хранения их в другом месте базы данных со ссылкой на конкретную ревизию.

Смотрите:

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. В итоге вы можете получить больше строк истории, чем вам нужно, но, по крайней мере, будет достаточно контекста, чтобы понять, что происходит.

Это решение регистрирует все команды в сессии, если были сделаны какие-либо изменения в базе данных.

Как обнаружить изменения в базе данных

Обертка 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

Проверьте, был ли использован 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)
Вернуться на верх