Add utility script for logging results of commands

src/charity_django/__init__.py

@@ -1 +1 @@
-__version__ = "0.18.0"
+__version__ = "0.19.0"

src/charity_django/utils/apps.py

@@ -0,0 +1,6 @@
+from django.apps import AppConfig
+
+
+class CharityUtilsConfig(AppConfig):
+    name = "charity_django.utils"
+    verbose_name = "Charity Utils"

src/charity_django/utils/management/commands/logcommand.py

@@ -0,0 +1,101 @@
+import datetime
+import logging
+import shlex
+from contextlib import redirect_stderr, redirect_stdout
+
+from charity_django.utils.models import CommandLog
+from django.core.management import call_command
+from django.core.management.base import BaseCommand
+
+
+class CommandLogHandler(logging.StreamHandler):
+    def __init__(self, commandlog):
+        logging.StreamHandler.__init__(self)
+        self.commandlog = commandlog
+        self.log = ""
+        self.errors = 0
+
+    def emit(self, record):
+        msg = self.format(record)
+        if not self.commandlog.log:
+            self.commandlog.log = ""
+        self.commandlog.log += msg + self.terminator
+        self.log += msg + self.terminator
+        if record.levelno in (logging.ERROR, logging.CRITICAL):
+            self.errors += 1
+        self.commandlog.save()
+        self.flush()
+
+    def teardown(self):
+        if self.errors > 0:
+            self.commandlog.status = CommandLog.CommandLogStatus.FAILED
+        else:
+            self.commandlog.status = CommandLog.CommandLogStatus.COMPLETED
+        if self.log and not self.commandlog.log:
+            self.commandlog.logs = self.log
+        self.commandlog.completed = datetime.datetime.now(datetime.timezone.utc)
+        self.commandlog.save()
+
+
+class LoggerWriter:
+    def __init__(self, level):
+        self.level = level
+
+    def write(self, message):
+        if isinstance(message, bytes):
+            message = message.decode("utf-8")
+        message = message.strip()
+        if message:
+            self.level(message)
+
+    def flush(self):
+        pass
+
+
+class Command(BaseCommand):
+    help = "Wrap a django command to save log messages"
+
+    def add_arguments(self, parser):
+        parser.add_argument("command", nargs="+")
+
+    def handle(self, *args, **options):
+        parsed_options = " ".join(options["command"]).split(" ", 1)
+        if len(parsed_options) == 1:
+            command, cmd_options = parsed_options[0], None
+        else:
+            command, cmd_options = parsed_options
+        self.stdout.write(f"Running command: {command}")
+        self.stdout.write(f"With options: {cmd_options}")
+
+        command_log = CommandLog.objects.create(
+            command=command,
+            cmd_options=cmd_options,
+            status=CommandLog.CommandLogStatus.RUNNING,
+            started=datetime.datetime.now(datetime.timezone.utc),
+        )
+
+        command_logger = CommandLogHandler(command_log)
+        command_log_format = logging.Formatter(
+            "{levelname} {asctime} [{name}] {message}", style="{"
+        )
+        command_logger.setFormatter(command_log_format)
+        command_logger.setLevel(logging.INFO)
+
+        logger = logging.getLogger()
+        logger.addHandler(command_logger)
+        logger.setLevel(logging.INFO)
+
+        try:
+            with redirect_stdout(LoggerWriter(logger.info)), redirect_stderr(
+                LoggerWriter(logger.error)
+            ):
+                if cmd_options:
+                    call_command(command, shlex.split(cmd_options))
+                else:
+                    call_command(command)
+        except Exception as err:
+            logger.exception(err)
+            command_logger.teardown()
+            raise
+
+        command_logger.teardown()

src/charity_django/utils/migrations/0001_initial.py

@@ -0,0 +1,43 @@
+# Generated by Django 5.0.4 on 2024-10-23 09:46
+
+from django.db import migrations, models
+
+
+class Migration(migrations.Migration):
+    initial = True
+
+    dependencies = []
+
+    operations = [
+        migrations.CreateModel(
+            name="CommandLog",
+            fields=[
+                (
+                    "id",
+                    models.BigAutoField(
+                        auto_created=True,
+                        primary_key=True,
+                        serialize=False,
+                        verbose_name="ID",
+                    ),
+                ),
+                ("command", models.CharField(max_length=255)),
+                ("cmd_options", models.TextField(blank=True, null=True)),
+                ("started", models.DateTimeField(auto_now_add=True)),
+                ("completed", models.DateTimeField(blank=True, null=True)),
+                (
+                    "status",
+                    models.IntegerField(
+                        choices=[
+                            (0, "Pending"),
+                            (1, "Running"),
+                            (2, "Completed"),
+                            (3, "Failed"),
+                        ],
+                        default=0,
+                    ),
+                ),
+                ("log", models.TextField(blank=True, null=True)),
+            ],
+        ),
+    ]

src/charity_django/utils/models.py

@@ -0,0 +1,18 @@
+from django.db import models
+
+
+class CommandLog(models.Model):
+    class CommandLogStatus(models.IntegerChoices):
+        PENDING = 0
+        RUNNING = 1
+        COMPLETED = 2
+        FAILED = 3
+
+    command = models.CharField(max_length=255)
+    cmd_options = models.TextField(null=True, blank=True)
+    started = models.DateTimeField(auto_now_add=True)
+    completed = models.DateTimeField(null=True, blank=True)
+    status = models.IntegerField(
+        choices=CommandLogStatus.choices, default=CommandLogStatus.PENDING
+    )
+    log = models.TextField(null=True, blank=True)

src/charity_django/utils/tests/test_utils_logcommand.py

@@ -0,0 +1,132 @@
+from charity_django.utils.models import CommandLog
+from django.core.management import call_command
+from django.test import TestCase
+
+
+class TestCommandLog(TestCase):
+    def test_success(self):
+        command_name = "test_command_success"
+        call_command("logcommand", command_name)
+        latest_log = CommandLog.objects.filter(command=command_name).latest("started")
+
+        self.assertEqual(latest_log.command, command_name)
+        self.assertEqual(latest_log.status, CommandLog.CommandLogStatus.COMPLETED)
+        self.assertIsNotNone(latest_log.completed)
+        self.assertTrue("Success stdout" in latest_log.log)
+        self.assertTrue("Success print" in latest_log.log)
+        self.assertTrue("Success log info" in latest_log.log)
+        self.assertTrue("Success log debug" not in latest_log.log)
+
+    def test_warning(self):
+        command_name = "test_command_warning"
+        call_command("logcommand", command_name)
+        latest_log = CommandLog.objects.filter(command=command_name).latest("started")
+
+        self.assertEqual(latest_log.command, command_name)
+        self.assertEqual(latest_log.status, CommandLog.CommandLogStatus.COMPLETED)
+        self.assertIsNotNone(latest_log.completed)
+        self.assertTrue("Success stdout" in latest_log.log)
+        self.assertTrue("Success print" in latest_log.log)
+        self.assertTrue("Success log info" in latest_log.log)
+        self.assertTrue("Success log debug" not in latest_log.log)
+        self.assertTrue("Success log warning" in latest_log.log)
+
+    def test_error(self):
+        command_name = "test_command_error"
+        call_command("logcommand", command_name)
+        latest_log = CommandLog.objects.filter(command=command_name).latest("started")
+
+        self.assertEqual(latest_log.command, command_name)
+        self.assertEqual(latest_log.status, CommandLog.CommandLogStatus.FAILED)
+        self.assertIsNotNone(latest_log.completed)
+        self.assertTrue("Success stdout" in latest_log.log)
+        self.assertTrue("Success print" in latest_log.log)
+        self.assertTrue("Success log info" in latest_log.log)
+        self.assertTrue("Success log debug" not in latest_log.log)
+        self.assertTrue("Success log warning" in latest_log.log)
+        self.assertTrue("Success log error" in latest_log.log)
+
+    def test_exception(self):
+        command_name = "test_command_exception"
+        error_raised = False
+        try:
+            call_command("logcommand", command_name)
+        except Exception:
+            error_raised = True
+
+        self.assertTrue(error_raised)
+
+        latest_log = CommandLog.objects.filter(command=command_name).latest("started")
+
+        self.assertEqual(latest_log.command, command_name)
+        self.assertEqual(latest_log.status, CommandLog.CommandLogStatus.FAILED)
+        self.assertIsNotNone(latest_log.completed)
+        self.assertTrue("Success stdout" in latest_log.log)
+        self.assertTrue("Success print" in latest_log.log)
+        # self.assertTrue("Success log info" in latest_log.log)
+        # self.assertTrue("Success log debug" not in latest_log.log)
+        # self.assertTrue("Success log warning" in latest_log.log)
+        # self.assertTrue("Success log error" in latest_log.log)
+        self.assertTrue("Error message" in latest_log.log)
+        self.assertTrue("LogCommandError" in latest_log.log)
+
+    def test_arguments(self):
+        command_name = "test_command_arguments"
+
+        args = [
+            ("--arg1 1 --arg2 2", ("arg1: 1", "arg2: 2")),
+            ("--arg1 1", ("arg1: 1", "arg2: None")),
+            ("", ("arg1: None", "arg2: None")),
+            ('--arg1 1 --arg2 "quoted string"', ("arg1: 1", "arg2: quoted string")),
+        ]
+        for argstr, expected_logs in args:
+            with self.subTest(argstr=argstr):
+                call_command("logcommand", f"{command_name} {argstr}")
+
+                latest_log = CommandLog.objects.filter(
+                    command=command_name, cmd_options=argstr
+                ).latest("started")
+
+                self.assertEqual(latest_log.command, command_name)
+                self.assertEqual(
+                    latest_log.status, CommandLog.CommandLogStatus.COMPLETED
+                )
+                self.assertIsNotNone(latest_log.completed)
+                self.assertTrue("Success stdout" in latest_log.log)
+                self.assertTrue("Success print" in latest_log.log)
+                self.assertTrue("Success log info" in latest_log.log)
+                self.assertTrue("Success log debug" not in latest_log.log)
+
+                for expected_log in expected_logs:
+                    self.assertTrue(expected_log in latest_log.log)
+
+    def test_arguments_exception(self):
+        command_name = "test_command_arguments"
+        argstr = "--arg1 99"
+        expected_logs = ("arg1: 99", "arg2: None")
+        error_raised = False
+        try:
+            call_command("logcommand", f"{command_name} {argstr}")
+        except Exception:
+            error_raised = True
+
+        self.assertTrue(error_raised)
+
+        latest_log = CommandLog.objects.filter(
+            command=command_name, cmd_options=argstr
+        ).latest("started")
+
+        self.assertEqual(latest_log.command, command_name)
+        self.assertEqual(latest_log.status, CommandLog.CommandLogStatus.FAILED)
+        self.assertIsNotNone(latest_log.completed)
+        self.assertTrue("Success stdout" in latest_log.log)
+        self.assertTrue("Success print" in latest_log.log)
+        # self.assertTrue("Success log info" in latest_log.log)
+        # self.assertTrue("Success log debug" not in latest_log.log)
+        # self.assertTrue("Success log warning" in latest_log.log)
+        # self.assertTrue("Success log error" in latest_log.log)
+        self.assertTrue("arg1 is 99" in latest_log.log)
+        self.assertTrue("LogCommandError" in latest_log.log)
+
+        for expected_log in expected_logs:
+            self.assertTrue(expected_log in latest_log.log)

test_project/config/settings.py

@@ -39,11 +39,13 @@
     "django.contrib.sessions",
     "django.contrib.messages",
     "django.contrib.staticfiles",
+    "test_project.test_app",
     "charity_django.ccew",
     "charity_django.oscr",
     "charity_django.ccni",
     "charity_django.companies",
     "charity_django.postcodes",
+    "charity_django.utils",
 ]
 
 MIDDLEWARE = [
@@ -104,6 +106,21 @@
 ]
 
 
+LOGGING = {
+    "version": 1,
+    "disable_existing_loggers": False,
+    "handlers": {
+        "console": {
+            "class": "logging.StreamHandler",
+        },
+    },
+    "root": {
+        "handlers": ["console"],
+        "level": "WARNING",
+    },
+}
+
+
 # Internationalization
 # https://docs.djangoproject.com/en/4.1/topics/i18n/
 

test_project/test_app/apps.py

@@ -0,0 +1,6 @@
+from django.apps import AppConfig
+
+
+class TestAppConfig(AppConfig):
+    name = "test_project.test_app"
+    verbose_name = "Test App"

test_project/test_app/management/commands/test_command_arguments.py

@@ -0,0 +1,31 @@
+import logging
+
+from django.core.management.base import BaseCommand
+
+from .test_command_exception import LogCommandError
+
+logger = logging.getLogger(__name__)
+
+
+class Command(BaseCommand):
+    help = (
+        "Test command which prints a message and completes successfully with arguments"
+    )
+
+    def add_arguments(self, parser):
+        parser.add_argument("--arg1", type=int, help="First argument")
+        parser.add_argument("--arg2", type=str, help="Second argument")
+
+    def handle(self, *args, **options):
+        self.stdout.write("Success stdout")
+        print("Success print")
+        logger.info("Success log info")
+        logger.debug("Success log debug")
+
+        arg1 = options["arg1"]
+        arg2 = options["arg2"]
+        print(f"arg1: {arg1}")
+        print(f"arg2: {arg2}")
+
+        if arg1 == 99:
+            raise LogCommandError("arg1 is 99")