bolt-jobs

Author: davegaeddert

bolt-jobs/.gitignore

@@ -0,0 +1,16 @@
+# Local development files
+/.env
+/.bolt
+*.sqlite3
+
+# Publishing
+/dist
+
+# Python
+/.venv
+__pycache__/
+*.py[cod]
+*$py.class
+
+# OS files
+.DS_Store

bolt-jobs/README.md

@@ -0,0 +1 @@
+# bolt-jobs

bolt-jobs/bolt/jobs/__init__.py

@@ -0,0 +1,3 @@
+from .jobs import Job
+
+__all__ = ["Job"]

bolt-jobs/bolt/jobs/admin.py

@@ -0,0 +1,153 @@
+from bolt.admin import (
+    AdminModelDetailView,
+    AdminModelListView,
+    AdminModelViewset,
+    register_viewset,
+)
+from bolt.admin.cards import Card
+from bolt.admin.dates import DatetimeRangeAliases
+from bolt.http import HttpResponseRedirect
+
+from .models import JobRequest, JobResult
+
+
+class SuccessfulJobsCard(Card):
+    title = "Successful Jobs"
+    text = "View"
+
+    def get_number(self):
+        return (
+            JobResult.objects.successful()
+            .filter(created_at__range=self.datetime_range.as_tuple())
+            .count()
+        )
+
+    def get_link(self):
+        return JobResultViewset.ListView.get_absolute_url() + "?filter=Successful"
+
+
+class ErroredJobsCard(Card):
+    title = "Errored Jobs"
+    text = "View"
+
+    def get_number(self):
+        return (
+            JobResult.objects.errored()
+            .filter(created_at__range=self.datetime_range.as_tuple())
+            .count()
+        )
+
+    def get_link(self):
+        return JobResultViewset.ListView.get_absolute_url() + "?filter=Errored"
+
+
+class LostJobsCard(Card):
+    title = "Lost Jobs"
+    text = "View"  # TODO make not required - just an icon?
+
+    def get_number(self):
+        return (
+            JobResult.objects.lost()
+            .filter(created_at__range=self.datetime_range.as_tuple())
+            .count()
+        )
+
+    def get_link(self):
+        return JobResultViewset.ListView.get_absolute_url() + "?filter=Lost"
+
+
+class RetriedJobsCard(Card):
+    title = "Retried Jobs"
+    text = "View"  # TODO make not required - just an icon?
+
+    def get_number(self):
+        return (
+            JobResult.objects.retried()
+            .filter(created_at__range=self.datetime_range.as_tuple())
+            .count()
+        )
+
+    def get_link(self):
+        return JobResultViewset.ListView.get_absolute_url() + "?filter=Retried"
+
+
+@register_viewset
+class JobRequestViewset(AdminModelViewset):
+    class ListView(AdminModelListView):
+        nav_section = "Jobs"
+        model = JobRequest
+        fields = ["id", "job_class", "priority", "created_at"]
+
+    class DetailView(AdminModelDetailView):
+        model = JobRequest
+
+
+@register_viewset
+class JobResultViewset(AdminModelViewset):
+    class ListView(AdminModelListView):
+        nav_section = "Jobs"
+        model = JobResult
+        fields = [
+            "id",
+            "job_class",
+            "priority",
+            "started_at",
+            "status",
+        ]
+        cards = [
+            SuccessfulJobsCard,
+            ErroredJobsCard,
+            LostJobsCard,
+            RetriedJobsCard,
+        ]
+        filters = [
+            "Successful",
+            "Processing",
+            "Errored",
+            "Lost",
+            "Retried",
+            "Unknown",
+        ]
+        actions = [
+            "Retry",
+        ]
+        allow_global_search = False
+        default_datetime_range = DatetimeRangeAliases.LAST_7_DAYS
+
+        def get_initial_queryset(self):
+            queryset = super().get_initial_queryset()
+            if self.filter == "Successful":
+                return queryset.successful()
+            if self.filter == "Errored":
+                return queryset.errored()
+            if self.filter == "Processing":
+                return queryset.processing()
+            if self.filter == "Lost":
+                return queryset.lost()
+            if self.filter == "Retried":
+                return queryset.retried()
+            if self.filter == "Unknown":
+                return queryset.unknown()
+            return queryset
+
+        def get_fields(self):
+            fields = super().get_fields()
+            if self.filter == "Retried":
+                fields.append("retries")
+                fields.append("retry_attempt")
+            return fields
+
+        def perform_action(self, action: str, target_pks: list):
+            if action == "Retry":
+                for result in JobResult.objects.filter(pk__in=target_pks):
+                    result.retry_job(delay=0)
+            else:
+                raise ValueError("Invalid action")
+
+    class DetailView(AdminModelDetailView):
+        model = JobResult
+
+        def post(self):
+            self.load_object()
+            self.object.retry_job(delay=0)
+            return HttpResponseRedirect(".")

bolt-jobs/bolt/jobs/cli.py

@@ -0,0 +1,73 @@
+import datetime
+import logging
+
+import click
+
+from bolt.utils import timezone
+
+from .models import JobRequest, JobResult
+from .workers import Worker
+
+logger = logging.getLogger("bolt.jobs")
+
+
+@click.group()
+def cli():
+    pass
+
+
+@cli.command()
+@click.option(
+    "--max-processes",
+    "max_processes",
+    default=None,
+    type=int,
+    envvar="BOLT_JOBS_MAX_PROCESSES",
+)
+@click.option(
+    "--max-jobs-per-process",
+    "max_jobs_per_process",
+    default=None,
+    type=int,
+    envvar="BOLT_JOBS_MAX_JOBS_PER_PROCESS",
+)
+@click.option(
+    "--stats-every",
+    "stats_every",
+    default=60,
+    type=int,
+    envvar="BOLT_JOBS_STATS_EVERY",
+)
+def worker(max_processes, max_jobs_per_process, stats_every):
+    Worker(
+        max_processes=max_processes,
+        max_jobs_per_process=max_jobs_per_process,
+        stats_every=stats_every,
+    ).run()
+
+
+@cli.command()
+@click.option("--older-than", type=int, default=60 * 60 * 24 * 7)
+def clear_completed(older_than):
+    cutoff = timezone.now() - datetime.timedelta(seconds=older_than)
+    click.echo(f"Clearing jobs finished before {cutoff}")
+    results = (
+        JobResult.objects.exclude(ended_at__isnull=True)
+        .filter(ended_at__lt=cutoff)
+        .delete()
+    )
+    click.echo(f"Deleted {results[0]} jobs")
+
+
+@cli.command()
+def stats():
+    pending = JobRequest.objects.count()
+
+    processing = JobResult.objects.processing().count()
+    successful = JobResult.objects.successful().count()
+    errored = JobResult.objects.errored().count()
+
+    click.echo(f"Pending: {click.style(pending, bold=True)}")
+    click.echo(f"Processing: {click.style(processing, bold=True)}")
+    click.echo(f"Successful: {click.style(successful, bold=True)}")
+    click.echo(f"Errored: {click.style(errored, bold=True)}")

bolt-jobs/bolt/jobs/config.py

@@ -0,0 +1,25 @@
+from importlib import import_module
+
+from bolt.packages import PackageConfig, packages
+
+MODULE_NAME = "jobs"
+
+
+class BoltJobsConfig(PackageConfig):
+    default_auto_field = "bolt.db.models.BigAutoField"
+    name = "bolt.jobs"
+    label = "boltqueue"
+
+    def ready(self):
+        # Trigger register calls to fire by importing the modules
+        for package_config in packages.get_package_configs():
+            try:
+                import_module(f"{package_config.name}.{MODULE_NAME}")
+            except ModuleNotFoundError:
+                pass
+
+        # Also trigger for the root app/admin.py module
+        try:
+            import_module(MODULE_NAME)
+        except ModuleNotFoundError:
+            pass

bolt-jobs/bolt/jobs/gid.py

@@ -0,0 +1,22 @@
+class GlobalID:
+    """A global identifier for a model instance."""
+
+    @staticmethod
+    def from_instance(instance):
+        return f"gid://{instance._meta.package_label}/{instance._meta.model_name}/{instance.pk}"
+
+    @staticmethod
+    def to_instance(s):
+        if not s.startswith("gid://"):
+            raise ValueError("Invalid GlobalID string")
+        package, model, pk = s[6:].split("/")
+        from bolt.packages import packages
+
+        model = packages.get_model(package, model)
+        return model.objects.get(pk=pk)
+
+    @staticmethod
+    def is_gid(x):
+        if not isinstance(x, str):
+            return False
+        return x.startswith("gid://")