Merge pull request #42 from dabapps/2.0-with-migrations

2.0 with migrations

Author: j4mie

.github/workflows/ci.yml

@@ -9,11 +9,12 @@ jobs:
 
     strategy:
       matrix:
-        python: [3.6, 3.7, 3.8]
-        django: [2.2]
+        python: [3.6, 3.7, 3.8, 3.9]
+        django: [3.1, 3.2]
         database_url:
           - postgres://runner:password@localhost/project
           - mysql://root:[email protected]/project
+          - 'sqlite:///:memory:'
 
     services:
       postgres:

.github/workflows/pypi.yml

@@ -0,0 +1,26 @@
+name: Upload Python Package
+
+on:
+  release:
+    types: [created]
+
+jobs:
+  deploy:
+    runs-on: ubuntu-latest
+    steps:
+    - uses: actions/checkout@v2
+    - name: Set up Python
+      uses: actions/setup-python@v2
+      with:
+        python-version: '3.8'
+    - name: Install dependencies
+      run: |
+        python -m pip install --upgrade pip
+        pip install setuptools wheel twine
+    - name: Build and publish
+      env:
+        TWINE_USERNAME: ${{ secrets.PYPI_USERNAME }}
+        TWINE_PASSWORD: ${{ secrets.PYPI_PASSWORD }}
+      run: |
+        python setup.py sdist bdist_wheel
+        twine upload dist/*

README.md

@@ -1,17 +1,14 @@
 django-db-queue
 ==========
-[![Build Status](https://travis-ci.org/dabapps/django-db-queue.svg)](https://travis-ci.org/dabapps/django-db-queue)
 [![pypi release](https://img.shields.io/pypi/v/django-db-queue.svg)](https://pypi.python.org/pypi/django-db-queue)
 
-Simple databased-backed job queue. Jobs are defined in your settings, and are processed by management commands.
+Simple database-backed job queue. Jobs are defined in your settings, and are processed by management commands.
 
 Asynchronous tasks are run via a *job queue*. This system is designed to support multi-step job workflows.
 
 Supported and tested against:
-- Django 1.11 and 2.2
-- Python 3.5, 3.6, 3.7 and 3.8
-
-This package may still work with older versions of Django and Python but they aren't explicitly supported.
+- Django 3.1 and 3.2
+- Python 3.6, 3.7, 3.8 and 3.9
 
 ## Getting Started
 
@@ -28,6 +25,10 @@ Add `django_dbq` to your installed apps
         'django_dbq',
     )
 
+### Upgrading from 1.x to 2.x
+
+Note that version 2.x only supports Django 3.1 or newer. If you need support for Django 2.2, please stick with the latest 1.x release.
+
 ### Describe your job
 
 In e.g. project.common.jobs:
@@ -209,15 +210,6 @@ jobs from the database which are in state `COMPLETE` or `FAILED` and were
 created more than 24 hours ago. This could be run, for example, as a cron task,
 to ensure the jobs table remains at a reasonable size.
 
-##### manage.py create_job
-For debugging/development purposes, a simple management command is supplied to create jobs:
-
-    manage.py create_job <job_name> --queue_name 'my_queue_name' --workspace '{"key": "value"}'
-
-The `workspace` flag is optional. If supplied, it must be a valid JSON string.
-
-`queue_name` is optional and defaults to `default`
-
 ##### manage.py worker
 To start a worker:
 

django_dbq/__init__.py

@@ -1 +1 @@
-__version__ = "1.3.1"
+__version__ = "2.0.0"

django_dbq/management/commands/create_job.py

@@ -3,9 +3,9 @@
 from django.utils import timezone
 from django.utils.module_loading import import_string
 from django_dbq.models import Job
-from simplesignals.process import WorkerProcessBase
 from time import sleep
 import logging
+import signal
 
 
 logger = logging.getLogger(__name__)
@@ -74,17 +74,27 @@ def process_job(queue_name):
         raise
 
 
-class Worker(WorkerProcessBase):
-
-    process_title = "jobworker"
-
+class Worker:
     def __init__(self, name, rate_limit_in_seconds):
         self.queue_name = name
         self.rate_limit_in_seconds = rate_limit_in_seconds
+        self.alive = True
         self.last_job_finished = None
-        super(Worker, self).__init__()
+        self.init_signals()
+
+    def init_signals(self):
+        signal.signal(signal.SIGINT, self.shutdown)
+        signal.signal(signal.SIGQUIT, self.shutdown)
+        signal.signal(signal.SIGTERM, self.shutdown)
+
+    def shutdown(self, signum, frame):
+        self.alive = False
+
+    def run(self):
+        while self.alive:
+            self.process_job()
 
-    def do_work(self):
+    def process_job(self):
         sleep(1)
         if (
             self.last_job_finished

django_dbq/management/commands/worker.py

@@ -2,7 +2,6 @@
 from __future__ import unicode_literals
 
 from django.db import models, migrations
-import jsonfield.fields
 import uuid
 
 from django.db.models import UUIDField
@@ -44,7 +43,7 @@ class Migration(migrations.Migration):
                     ),
                 ),
                 ("next_task", models.CharField(max_length=100, blank=True)),
-                ("workspace", jsonfield.fields.JSONField(null=True)),
+                ("workspace", models.TextField(null=True)),
                 (
                     "queue_name",
                     models.CharField(db_index=True, max_length=20, default="default"),

django_dbq/migrations/0001_initial.py

@@ -0,0 +1,32 @@
+# Generated by Django 3.2rc1 on 2021-08-18 02:47
+
+from django.db import migrations, models
+
+
+class Migration(migrations.Migration):
+
+    dependencies = [
+        ("django_dbq", "0003_auto_20180713_1000"),
+    ]
+
+    operations = [
+        migrations.AlterField(
+            model_name="job",
+            name="state",
+            field=models.CharField(
+                choices=[
+                    ("NEW", "New"),
+                    ("READY", "Ready"),
+                    ("PROCESSING", "Processing"),
+                    ("FAILED", "Failed"),
+                    ("COMPLETE", "Complete"),
+                ],
+                db_index=True,
+                default="NEW",
+                max_length=20,
+            ),
+        ),
+        migrations.AlterField(
+            model_name="job", name="workspace", field=models.JSONField(null=True),
+        ),
+    ]

django_dbq/migrations/0004_auto_20210818_0247.py

@@ -6,8 +6,7 @@
     get_failure_hook_name,
     get_creation_hook_name,
 )
-from jsonfield import JSONField
-from django.db.models import UUIDField, Count
+from django.db.models import JSONField, UUIDField, Count, TextChoices
 import datetime
 import logging
 import uuid
@@ -74,27 +73,19 @@ def to_process(self, queue_name):
 
 
 class Job(models.Model):
-    class STATES:
+    class STATES(TextChoices):
         NEW = "NEW"
         READY = "READY"
         PROCESSING = "PROCESSING"
         FAILED = "FAILED"
         COMPLETE = "COMPLETE"
 
-    STATE_CHOICES = [
-        (STATES.NEW, "NEW"),
-        (STATES.READY, "READY"),
-        (STATES.PROCESSING, "PROCESSING"),
-        (STATES.FAILED, "FAILED"),
-        (STATES.COMPLETE, "COMPLETE"),
-    ]
-
     id = UUIDField(primary_key=True, default=uuid.uuid4, editable=False)
     created = models.DateTimeField(auto_now_add=True, db_index=True)
     modified = models.DateTimeField(auto_now=True)
     name = models.CharField(max_length=100)
     state = models.CharField(
-        max_length=20, choices=STATE_CHOICES, default=STATES.NEW, db_index=True
+        max_length=20, choices=STATES.choices, default=STATES.NEW, db_index=True
     )
     next_task = models.CharField(max_length=100, blank=True)
     workspace = JSONField(null=True)
@@ -107,9 +98,7 @@ class Meta:
     objects = JobManager()
 
     def save(self, *args, **kwargs):
-        is_new = not Job.objects.filter(pk=self.pk).exists()
-
-        if is_new:
+        if self._state.adding:
             self.next_task = get_next_task_name(self.name)
             self.workspace = self.workspace or {}
 
@@ -121,7 +110,7 @@ def save(self, *args, **kwargs):
                 )
                 return  # cancel the save
 
-        return super(Job, self).save(*args, **kwargs)
+        return super().save(*args, **kwargs)
 
     def update_next_task(self):
         self.next_task = get_next_task_name(self.name, self.next_task) or ""