add abstract model

.github/workflows/test.yml

@@ -61,7 +61,7 @@ jobs:
     strategy:
       max-parallel: 1
       matrix:
-        python-version: [ "3.11", "3.12" ]
+        python-version: [ "3.11", "3.12", "3.13" ]
         django-version: [ "4.2", "5.1" ]
       fail-fast: true
     needs: [ changes ]

pyproject.toml

@@ -9,6 +9,7 @@ dependencies = [
     "django-admin-extra-buttons>=1.5.8",
     "django-concurrency>=2.6",
     "django>=4",
+    "sentry-sdk",
 ]
 
 [project.optional-dependencies]

src/django_celery_boost/models.py

@@ -3,11 +3,13 @@
 import logging
 from typing import TYPE_CHECKING, Any, Callable, Generator, Optional
 
+import sentry_sdk
 from celery import states
 from celery.app.base import Celery
 from celery.result import AsyncResult
 from concurrency.api import concurrency_disable_increment
 from concurrency.fields import AutoIncVersionField
+from django.apps import apps
 from django.conf import settings
 from django.core import checks
 from django.db import models
@@ -377,3 +379,60 @@ def purge(cls: "type[CeleryTaskModel]") -> None:
         with cls.celery_app.pool.acquire(block=True) as conn:
             conn.default_channel.client.delete(cls.celery_task_queue)
             conn.default_channel.client.delete(cls.celery_task_revoked_queue)
+
+
+class AsyncJob(CeleryTaskModel):
+    class JobType(models.TextChoices):
+        STANDARD_TASK = "STANDARD_TASK", "Standard Task"
+        ADMIN_ACTION = "ADMIN_ACTION", "Admin Action"
+        JOB_TASK = "JOB_TASK", "Job Task"
+
+    type = models.CharField(max_length=50, choices=JobType.choices)
+    config = models.JSONField(default=dict, blank=True)
+    action = models.CharField(max_length=500, blank=True, null=True)
+    description = models.CharField(max_length=255, blank=True, null=True)
+    sentry_id = models.CharField(max_length=255, blank=True, null=True)
+
+    class Meta:
+        abstract = True
+        permissions = (("debug_job", "Can debug background jobs"),)
+
+    def __str__(self):
+        return self.description or f"Background Job #{self.pk}"
+
+    @property
+    def queue_position(self) -> int:
+        try:
+            return super().queue_position
+        except Exception:
+            return 0
+
+    @property
+    def started(self) -> str:
+        try:
+            return self.task_info["started_at"]
+        except Exception:
+            return "="
+
+    def execute(self):
+        sid = None
+        try:
+            func = import_string(self.action)
+            match self.type:
+                case AsyncJob.JobType.STANDARD_TASK:
+                    return func(**self.config)
+                case AsyncJob.JobType.ADMIN_ACTION:
+                    model = apps.get_model(self.config["model_name"])
+                    qs = model.objects.all()
+                    if self.config["pks"] != "__all__":
+                        qs = qs.filter(pk__in=self.config["pks"])
+                    return func(qs, **self.config.get("kwargs", {}))
+                case AsyncJob.JobType.JOB_TASK:
+                    return func(self)
+        except Exception as e:
+            sid = sentry_sdk.capture_exception(e)
+            raise e
+        finally:
+            if sid:
+                self.sentry_id = sid
+                self.save(update_fields=["sentry_id"])

tests/demoapp/demo/migrations/0001_initial.py

@@ -1,4 +1,4 @@
-# Generated by Django 5.1.3 on 2024-11-28 14:38
+# Generated by Django 5.1.1 on 2024-12-17 16:42
 
 import concurrency.fields
 import django.db.models.deletion
@@ -153,4 +153,81 @@ class Migration(migrations.Migration):
                 "default_permissions": ("add", "change", "delete", "view", "queue", "terminate", "inspect", "revoke"),
             },
         ),
+        migrations.CreateModel(
+            name="MultipleJob",
+            fields=[
+                ("id", models.BigAutoField(auto_created=True, primary_key=True, serialize=False, verbose_name="ID")),
+                ("version", concurrency.fields.AutoIncVersionField(default=0, help_text="record revision number")),
+                (
+                    "curr_async_result_id",
+                    models.CharField(
+                        blank=True,
+                        editable=False,
+                        help_text="Current (active) AsyncResult is",
+                        max_length=36,
+                        null=True,
+                    ),
+                ),
+                (
+                    "last_async_result_id",
+                    models.CharField(
+                        blank=True, editable=False, help_text="Latest executed AsyncResult is", max_length=36, null=True
+                    ),
+                ),
+                ("datetime_created", models.DateTimeField(auto_now_add=True, help_text="Creation date and time")),
+                (
+                    "datetime_queued",
+                    models.DateTimeField(
+                        blank=True, help_text="Queueing date and time", null=True, verbose_name="Queued At"
+                    ),
+                ),
+                (
+                    "repeatable",
+                    models.BooleanField(
+                        blank=True, default=False, help_text="Indicate if the job can be repeated as-is"
+                    ),
+                ),
+                ("celery_history", models.JSONField(blank=True, default=dict, editable=False)),
+                ("local_status", models.CharField(blank=True, default="", editable=False, max_length=100, null=True)),
+                (
+                    "group_key",
+                    models.CharField(
+                        blank=True,
+                        editable=False,
+                        help_text="Tasks with the same group key will not run in parallel",
+                        max_length=255,
+                        null=True,
+                    ),
+                ),
+                (
+                    "type",
+                    models.CharField(
+                        choices=[
+                            ("STANDARD_TASK", "Standard Task"),
+                            ("ADMIN_ACTION", "Admin Action"),
+                            ("JOB_TASK", "Job Task"),
+                        ],
+                        max_length=50,
+                    ),
+                ),
+                ("config", models.JSONField(blank=True, default=dict)),
+                ("action", models.CharField(blank=True, max_length=500, null=True)),
+                ("description", models.CharField(blank=True, max_length=255, null=True)),
+                ("sentry_id", models.CharField(blank=True, max_length=255, null=True)),
+                (
+                    "owner",
+                    models.ForeignKey(
+                        blank=True,
+                        null=True,
+                        on_delete=django.db.models.deletion.CASCADE,
+                        related_name="%(app_label)s_%(class)s_jobs",
+                        to=settings.AUTH_USER_MODEL,
+                    ),
+                ),
+            ],
+            options={
+                "permissions": (("test_multiplejob", "Can test MultipleJob"),),
+                "abstract": False,
+            },
+        ),
     ]

tests/demoapp/demo/models.py

@@ -1,6 +1,6 @@
 from django.db import models
 
-from django_celery_boost.models import CeleryTaskModel
+from django_celery_boost.models import AsyncJob, CeleryTaskModel
 
 
 class Job(CeleryTaskModel, models.Model):
@@ -31,3 +31,10 @@ class Meta(CeleryTaskModel.Meta):
         permissions = (("test_alternativejob", "Can test AlternativeJob"),)
 
     celery_task_name = "demo.tasks.process_job"
+
+
+class MultipleJob(AsyncJob):
+    class Meta(AsyncJob.Meta):
+        permissions = (("test_multiplejob", "Can test MultipleJob"),)
+
+    celery_task_name = "demo.tasks.process_job"

tests/test_permissions.py

@@ -6,3 +6,4 @@ def test_permissions_created(db):
     assert Permission.objects.filter(content_type__app_label="demo", codename="test_alternativejob").exists()
 
     assert Permission.objects.filter(content_type__app_label="demo", codename="queue_alternativejob").exists()
+    assert Permission.objects.filter(content_type__app_label="demo", codename="test_multiplejob").exists()

tox.ini

@@ -1,5 +1,5 @@
 [tox]
-envlist = d{42,51}-py{311,312}
+envlist = d{42,51}-py{311,312,313}
 skip_missing_interpreters = true
 
 [testenv]