Merge pull request #157 from WikiMovimentoBrasil/2025-01-improve-background-runners

feat: improve background runners and other stuff

Author: mgalves

src/app.ini

@@ -15,3 +15,5 @@ vacuum = true
 die-on-term = true
 
 attach-daemon = python3 manage.py send_batches
+attach-daemon = python3 manage.py send_batches
+attach-daemon = python3 manage.py send_batches

src/core/management/commands/send_batches.py

@@ -1,7 +1,12 @@
 import time
+import logging
 
 from django.core.management.base import BaseCommand
 from core.models import Batch
+from django.db import transaction
+from django.db.utils import OperationalError
+
+logger = logging.getLogger("qsts3")
 
 
 class Command(BaseCommand):
@@ -13,20 +18,34 @@ def handle(self, *args, **options):
         self.send_start_message()
 
         while True:
-            batch = self.get_first_batch()
+            batch = self.get_first_batch_and_lock()
             if batch:
                 batch.run()
             else:
                 time.sleep(self.TIMEOUT_SEC)
 
     def send_start_message(self):
-        msg = self.style.SUCCESS("==> send_batches management command started!")
-        self.stdout.write(msg)
+        logger.info("[command] send_batches management command started!")
+
+    def get_first_batch_and_lock(self):
+        with transaction.atomic():
+            try:
+                batch = self.get_first_batch()
+                self.lock_batch(batch)
+            except OperationalError:
+                logger.debug("[command] race condition mitigated")
+                return None
+        return batch
 
     def get_first_batch(self):
         return (
-            Batch.objects.select_for_update()
+            Batch.objects.select_for_update(skip_locked=True)
             .filter(status=Batch.STATUS_INITIAL)
             .order_by("id")
             .first()
         )
+
+    def lock_batch(self, batch):
+        if batch:
+            batch.status = batch.STATUS_RUNNING
+            batch.save()

src/core/models.py

@@ -85,8 +85,8 @@ def run(self):
         Sends all the batch commands to the Wikidata API. This method should not fail.
         Sets the batch status to BLOCKED when a command fails.
         """
-        # Ignore when not INITIAL
-        if not self.is_initial:
+        # Ignore when not INITIAL or RUNNING
+        if not self.is_initial_or_running:
             return
 
         self.start()
@@ -146,10 +146,13 @@ def finish(self):
         self.save()
 
     def stop(self):
-        logger.debug(f"[{self}] stop...")
-        self.message = f"Batch stopped processing by owner at {datetime.now()}"
-        self.status = self.STATUS_STOPPED
-        self.save()
+        if not self.is_done:
+            logger.debug(f"[{self}] stop...")
+            self.message = f"Batch stopped processing by owner at {datetime.now()}"
+            self.status = self.STATUS_STOPPED
+            self.save()
+        else:
+            logger.debug(f"[{self}] user tried to stop but batch is done.")
 
     def restart(self):
         if self.is_stopped:
@@ -386,7 +389,7 @@ class Error(models.TextChoices):
     )
 
     def __str__(self):
-        return f"Batch #{self.batch.pk} Command #{self.pk}"
+        return f"Batch #{self.batch.pk} Command #{self.pk} ##{self.index}"
 
     # -----------------
     # Status-changing methods
@@ -528,13 +531,21 @@ def parser_value_to_api_value(self, parser_value):
 
     @property
     def statement_api_value(self):
+        self.update_quantity_units_if_needed()
+        return self.parser_value_to_api_value(self.json["value"])
+
+    def update_quantity_units_if_needed(self):
+        # TODO: maybe we can add this elsewhere,
+        # because `statement_api_value` above is called a lot
         value = self.json["value"]
-        if value["type"] == "quantity" and value["value"]["unit"] != "1":
-            base = self.batch.wikibase_url()
+        base = self.batch.wikibase_url()
+        if (value["type"] == "quantity"
+            and value["value"]["unit"] != "1"
+            and base not in value["value"]["unit"]
+        ):
             unit_id = value["value"]["unit"]
             full_unit = f"{base}/entity/Q{unit_id}"
             value["value"]["unit"] = full_unit
-        return self.parser_value_to_api_value(value)
 
     def update_statement(self, st):
         st.setdefault("property", {"id": self.prop})
@@ -787,7 +798,7 @@ def update_combining_state(self, client: Client):
             commands=commands,
             entity=entity,
         )
-        logger.debug(f"[{self}] combined. final entity={entity}")
+        logger.debug(f"[{self}] combined with next")
 
     @property
     def final_combining_state(self):

src/qsts3/settings.py

@@ -141,7 +141,7 @@
     "version": 1,
     "disable_existing_loggers": False,
     "formatters": {
-        "simple": {"format": "%(asctime)s [%(levelname)s] %(message)s"},
+        "simple": {"format": "%(asctime)s [pid=%(process)d] [%(levelname)s] %(message)s"},
         "complete": {"format": "%(asctime)s - [%(levelname)s] %(name)s => %(message)s"},
     },
     "handlers": {

src/web/templates/batch.html

@@ -41,12 +41,8 @@ <h2> {% translate "Batch" %} #{{ batch.pk }} <img id="spinner" class="htmx-indic
 
 <div id="batchProgressDiv" 
       hx-get="{% url 'batch_summary' pk=batch.pk %}"
-      {% if batch.is_initial_or_running %}
-      hx-trigger="load, every 3s"
-      {% else %}
       hx-trigger="load"
-      {% endif %}
-      hx-swap="innerHTML" 
+      hx-swap="outerHTML"
       hx-indicator="#spinner"
       style="margin: 20px 0; font-size: 14px;">
       {% translate "Loading summary..." %}

src/web/templates/batch_summary.html

@@ -1,5 +1,14 @@
 {% load i18n %}
 
+<div id="batchProgressDiv"
+      {% if batch.is_initial_or_running %}
+      hx-get="{% url 'batch_summary' pk=pk %}?previous_status={{ batch.status }}"
+      hx-trigger="load delay:3s"
+      hx-swap="outerHTML"
+      hx-indicator="#spinner"
+      {% endif %}
+      style="margin: 20px 0; font-size: 14px;">
+
 <div class="progress" id="batchprogress">
   <div class="progress-status">{% translate "BATCH STATUS" %}
     <b class="status status_{{ status | lower }}">{{status | upper}}</b>
@@ -31,3 +40,5 @@
     </small>
     {% endif %}
 </div>
+
+</div>

src/web/templates/preview_batch.html

@@ -30,7 +30,7 @@ <h2> {% translate "Batch Preview" %} <img id="spinner" class="htmx-indicator"></
   </p>
 </div>
 <div id="batchProgressDiv" style="margin: 20px 0; font-size: 14px; ">
-    {% include 'batch_summary.html' with done_count=0 done_percentage=0 status="Preview" show_block_on_errors_notice=batch.block_on_errors %}
+    {% include 'batch_summary.html' with batch=batch done_count=0 done_percentage=0 status="Preview" show_block_on_errors_notice=batch.block_on_errors %}
 </div>
 <h4>{% translate "Commands" %}</h4>
 <div class="overflow-auto" id="batchCommandsDiv" hx-get="{% url 'preview_batch_commands' %}" hx-trigger="load, reload" hx-indicator="#spinner" hx-swap="innerHTML">

src/web/tests/test_views.py

@@ -598,7 +598,7 @@ def test_batch_report(self, mocker):
         result = (
             """b'batch_id,index,operation,status,error,message,entity_id,raw_input,api_response\\r\\n"""
             """1,0,set_statement,Done,,,Q1234,Q1234|P2|Q1,{\\\'id\\\': \\\'Q1234$stuff\\\'}\\r\\n"""
-            """1,1,set_label,Done,,,Q11,"Q11|Len|""label"" ","""
+            """1,1,set_label,Done,,,Q11,"Q11|Len|""label""\","""
             """"{\\\'id\\\': \\\'Q11\\\', \\\'labels\\\': {\\\'en\\\': \\\'label\\\'}}"\\r\\n\'"""
         )
         self.assertEqual(result, str(response.content).strip())

src/web/views/batch.py

@@ -174,11 +174,12 @@ def batch_summary(request, pk):
         )
         show_block_on_errors_notice = batch.is_preview_initial_or_running and batch.block_on_errors
 
-        return render(
+        response = render(
             request,
             "batch_summary.html",
             {
                 "pk": batch.pk,
+                "batch": batch,
                 "status": batch.get_status_display(),
                 "error_count": batch.error_commands,
                 "initial_count": batch.initial_commands,
@@ -191,5 +192,12 @@ def batch_summary(request, pk):
                 "show_block_on_errors_notice": show_block_on_errors_notice,
             },
         )
+        if batch.is_done:
+            previous_status = request.GET.get("previous_status")
+            if previous_status and previous_status != str(batch.STATUS_DONE):
+                # Refreshing the page to load the correct buttons
+                # and reload the command list if it's the first DONE
+                response.headers["HX-Refresh"] = "true"
+        return response
     except Batch.DoesNotExist:
         return render(request, "batch_summary.html", {}, status=404)

src/web/views/new_batch.py

@@ -147,6 +147,7 @@ def new_batch(request):
 
             batch = parser.parse(batch_name, batch_owner, batch_commands)
 
+            batch.status = batch.STATUS_PREVIEW
             batch.block_on_errors = "block_on_errors" in request.POST
             batch.combine_commands = "do_not_combine_commands" not in request.POST