Me gustaría generar automáticamente algún tipo de registro de todos los cambios en la base de datos que se realizan a través del shell de Django en el entorno de producción.
Utilizamos secuencias de comandos de migración de datos y esquemas para modificar la base de datos de producción y están controlados por versión. Por lo tanto, si introducimos un error, es fácil rastrearlo. Pero si un desarrollador del equipo cambia la base de datos a través del shell de Django, lo que luego presenta un problema, por el momento solo podemos esperar que recuerden lo que hicieron o que podamos encontrar sus comandos en el historial del shell de Python.
Ejemplo. Imaginemos que el siguiente código fue ejecutado por un desarrollador del equipo a través del shell de Python:
>>> tm = TeamMembership.objects.get(person=alice) >>> tm.end_date = date(2022,1,1) >>> tm.save()Cambia un objeto de membresía de equipo en la base de datos. Me gustaría registrar esto de alguna manera.
Soy consciente de que hay un montón de paquetes de Django relacionados con el registro de auditoría , pero solo estoy interesado en los cambios que se activan desde el shell de Django, y quiero registrar el código de Python que actualizó los datos.
Así que las preguntas que tengo en mente:
pre_save para todos los modelos para saber si los datos cambian, pero ¿cómo puedo saber si la fuente fue del shell de Python? ¿Cómo sé cuál era la declaración original de Python?Podrías usar la anotación del receiver de Django.
Por ejemplo, si desea detectar alguna llamada del método save , podría hacer:
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")algo más de documentación para las señales
Yo consideraría algo como esto:
Envolviendo cada sesión de python con algún tipo de código de inicialización usando, por ejemplo, la variable de entorno PYTHONSTARTUP https://docs.python.org/3/using/cmdline.html#envvar-PYTHONSTARTUP
En el archivo donde PYTHONSTARTUP apunta al registro del controlador de salida usando atexit https://docs.python.org/3/library/atexit.html
Estas dos cosas deberían permitirle usar algunas API de nivel inferior de django-reversion para envolver toda la sesión de terminal con https://django-reversion.readthedocs.io/en/stable/api.html#creating-revisions (algo como esto pero llamando a __enter__ y __exit__ de ese administrador de contexto directamente en su código de inicio y atexit ). Desafortunadamente, no conozco los detalles, pero debería ser factible.
En atexit /revision, finalice llamando al código para enumerar las líneas adicionales de la sesión de terminal y almacenándolas en otro lugar de la base de datos con una referencia a la revisión específica.
Ver:
https://docs.python.org/3/library/readline.html#readline.get_history_length
https://docs.python.org/3/library/readline.html#readline.get_history_item
Básicamente, la idea es que podría llamar a get_history_length dos veces: al principio y al final de la sesión de terminal. Eso le permitirá obtener líneas relevantes de dónde tuvo lugar el cambio usando get_history_item . Puede terminar teniendo más líneas de historial de las que realmente necesita, pero al menos hay suficiente contexto para ver qué está pasando.
Esta solución registra todos los comandos en la sesión si se realizaron cambios en la base de datos.
Envuelva execute_sql de SQLInsertCompiler , SQLUpdateCompiler y SQLDeleteCompiler .
SQLDeleteCompiler.execute_sql devuelve un contenedor de cursor.
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) atexit.register() un controlador de salida que hace readline.write_history_file() .
import atexit import readline def exit_handler(): filename = 'history.py' readline.write_history_file(filename) atexit.register(exit_handler) Compruebe si se utilizó IPython comparando 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) Agregue lo siguiente en manage.py antes 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)Basado en la respuesta de aaron y la implementación del %logstart mágico incorporado de IPython , esta es la solución que encontramos al final.
Todos los comandos de la última sesión de IPython se registran en un archivo de historial si alguno de los comandos activó una escritura en la base de datos a través del ORM de Django.
Aquí hay un extracto del archivo de historial generado:
❯ 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() Aquí está nuestro manage.py ahora:
#!/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)