Support for Django 5.1 & Python 3.12, and removed the django-common dep

.github/workflows/django.yml

@@ -1,33 +1,39 @@
-name: Django CI
+name: CI
+permissions: read-all
 
 on:
   push:
-    branches: [ master ]
+    branches:
+      - master
   pull_request:
-    branches: [ master ]
+  workflow_dispatch:
+
+concurrency:
+  group: ${{ github.workflow }}-${{ github.head_ref || github.run_id }}
+  cancel-in-progress: true
+
+env:
+  UV_LOCKED: 1
 
 jobs:
   build:
-
     runs-on: ubuntu-latest
+    timeout-minutes: 10
+
     strategy:
-      max-parallel: 4
+      fail-fast: false
       matrix:
-        python-version: [3.7, 3.8, 3.9]
+        python: ["3.10", "3.11", "3.12", "3.13", "3.14"]
 
     steps:
-    - uses: actions/checkout@v3
-    - name: Set up Python ${{ matrix.python-version }}
-      uses: actions/setup-python@v3
-      with:
-        python-version: ${{ matrix.python-version }}
-    - name: Install Dependencies
-      run: |
-        python -m pip install --upgrade pip
-        pip install .
-    - name: Install Test Dependencies
-      run: |
-        pip install -r test_requirements.txt
-    - name: Run Tests
-      run: |
-        python runtests.py runtests
+      - uses: actions/[email protected]
+
+      - uses: astral-sh/[email protected]
+        with:
+          python-version: ${{ matrix.python }}
+
+      - name: django-admin check
+        run: uv run testmanage.py check
+
+      - name: django-admin test
+        run: uv run testmanage.py test

django_cron/__init__.py

@@ -1,2 +1,17 @@
-from django_cron.core import *
-from django_cron.helpers import get_class, get_current_time
+from .core import (
+    DEFAULT_LOCK_BACKEND,
+    DJANGO_CRON_OUTPUT_ERRORS,
+    BadCronJobError,
+    CronJobBase,
+    CronJobManager,
+    Schedule,
+)
+
+__all__ = (
+    "DEFAULT_LOCK_BACKEND",
+    "DJANGO_CRON_OUTPUT_ERRORS",
+    "BadCronJobError",
+    "CronJobBase",
+    "CronJobManager",
+    "Schedule",
+)

django_cron/admin.py

@@ -4,59 +4,62 @@
 from django.db.models import DurationField, ExpressionWrapper, F
 from django.utils.translation import gettext_lazy as _
 
-from django_cron.models import CronJobLog, CronJobLock
-from django_cron.helpers import humanize_duration
+from .helpers import humanize_duration
+from .models import CronJobLock, CronJobLog
 
 
 class DurationFilter(admin.SimpleListFilter):
-    title = _('duration')
-    parameter_name = 'duration'
+    title = _("duration")
+    parameter_name = "duration"
 
     def lookups(self, request, model_admin):
         return (
-            ('lte_minute', _('<= 1 minute')),
-            ('gt_minute', _('> 1 minute')),
-            ('gt_hour', _('> 1 hour')),
-            ('gt_day', _('> 1 day')),
+            ("lte_minute", _("<= 1 minute")),
+            ("gt_minute", _("> 1 minute")),
+            ("gt_hour", _("> 1 hour")),
+            ("gt_day", _("> 1 day")),
         )
 
     def queryset(self, request, queryset):
-        if self.value() == 'lte_minute':
-            return queryset.filter(end_time__lte=F('start_time') + timedelta(minutes=1))
-        if self.value() == 'gt_minute':
-            return queryset.filter(end_time__gt=F('start_time') + timedelta(minutes=1))
-        if self.value() == 'gt_hour':
-            return queryset.filter(end_time__gt=F('start_time') + timedelta(hours=1))
-        if self.value() == 'gt_day':
-            return queryset.filter(end_time__gt=F('start_time') + timedelta(days=1))
-
+        if self.value() == "lte_minute":
+            return queryset.filter(end_time__lte=F("start_time") + timedelta(minutes=1))
+        if self.value() == "gt_minute":
+            return queryset.filter(end_time__gt=F("start_time") + timedelta(minutes=1))
+        if self.value() == "gt_hour":
+            return queryset.filter(end_time__gt=F("start_time") + timedelta(hours=1))
+        if self.value() == "gt_day":
+            return queryset.filter(end_time__gt=F("start_time") + timedelta(days=1))
 
+@admin.register(CronJobLog)
 class CronJobLogAdmin(admin.ModelAdmin):
     class Meta:
         model = CronJobLog
 
-    search_fields = ('code', 'message')
-    ordering = ('-start_time',)
-    list_display = ('code', 'start_time', 'end_time', 'humanize_duration', 'is_success')
-    list_filter = ('code', 'start_time', 'is_success', DurationFilter)
+    search_fields = ("code", "message")
+    ordering = ("-start_time",)
+    list_display = ("code", "start_time", "end_time", "humanize_duration", "is_success")
+    list_filter = ("code", "start_time", "is_success", DurationFilter)
 
     def get_queryset(self, request):
-        return super().get_queryset(request).annotate(
-            duration=ExpressionWrapper(F('end_time') - F('start_time'), DurationField()),
+        return (
+            super()
+            .get_queryset(request)
+            .annotate(
+                duration=ExpressionWrapper(
+                    F("end_time") - F("start_time"), DurationField()
+                ),
+            )
         )
 
     def get_readonly_fields(self, request, obj=None):
-        if not request.user.is_superuser and obj is not None:
-            names = [f.name for f in CronJobLog._meta.fields if f.name != 'id']
-            return self.readonly_fields + tuple(names)
+        if obj is not None and not getattr(request.user, "is_superuser", False):
+            names = [f.name for f in CronJobLog._meta.fields if f.name != "id"]
+            return *self.readonly_fields, *names
         return self.readonly_fields
 
+    @admin.display(description=_("Duration"), ordering="duration")
     def humanize_duration(self, obj):
         return humanize_duration(obj.end_time - obj.start_time)
 
-    humanize_duration.short_description = _("Duration")
-    humanize_duration.admin_order_field = 'duration'
-
 
-admin.site.register(CronJobLog, CronJobLogAdmin)
 admin.site.register(CronJobLock)

django_cron/backends/lock/base.py

@@ -1,4 +1,4 @@
-class DjangoCronJobLock(object):
+class DjangoCronJobLock:
     """
     The lock class to use in runcrons management command.
     Intendent usage is
@@ -12,10 +12,15 @@ class DjangoCronJobLock(object):
         pass
     """
 
+    job_name: str
+    job_code: str
+    parallel: bool
+    silent: bool
+
     class LockFailedException(Exception):
         pass
 
-    def __init__(self, cron_class, silent, *args, **kwargs):
+    def __init__(self, cron_class: type, silent: bool, *args, **kwargs):
         """
         This method inits the class.
         You should take care of getting all
@@ -27,12 +32,12 @@ def __init__(self, cron_class, silent, *args, **kwargs):
             * self.silent
         for you. The rest is backend-specific.
         """
-        self.job_name = '.'.join([cron_class.__module__, cron_class.__name__])
+        self.job_name = f"{cron_class.__module__}.{cron_class.__name__}"
         self.job_code = cron_class.code
-        self.parallel = getattr(cron_class, 'ALLOW_PARALLEL_RUNS', False)
+        self.parallel = getattr(cron_class, "ALLOW_PARALLEL_RUNS", False)
         self.silent = silent
 
-    def lock(self):
+    def lock(self) -> bool:
         """
         This method called to acquire lock. Typically. it will
         be called from __enter__ method.
@@ -41,26 +46,26 @@ def lock(self):
         Here you can optionally call self.notice_lock_failed().
         """
         raise NotImplementedError(
-            'You have to implement lock(self) method for your class'
+            "You have to implement lock(self) method for your class"
         )
 
-    def release(self):
+    def release(self) -> None:
         """
         This method called to release lock.
         Tipically called from __exit__ method.
         No need to return anything currently.
         """
         raise NotImplementedError(
-            'You have to implement release(self) method for your class'
+            "You have to implement release(self) method for your class"
         )
 
-    def lock_failed_message(self):
+    def lock_failed_message(self) -> str | list[str]:
         return "%s: lock found. Will try later." % self.job_name
 
-    def __enter__(self):
+    def __enter__(self) -> None:
         if not self.parallel and not self.lock():
             raise self.LockFailedException(self.lock_failed_message())
 
-    def __exit__(self, type, value, traceback):
+    def __exit__(self, type, value, traceback) -> None:
         if not self.parallel:
             self.release()

django_cron/backends/lock/cache.py

@@ -1,8 +1,8 @@
 from django.conf import settings
-from django.core.cache import caches
+from django.core.cache import BaseCache, caches
 from django.utils import timezone
 
-from django_cron.backends.lock.base import DjangoCronJobLock
+from .base import DjangoCronJobLock
 
 
 class CacheLock(DjangoCronJobLock):
@@ -13,7 +13,11 @@ class CacheLock(DjangoCronJobLock):
 
     DEFAULT_LOCK_TIME = 24 * 60 * 60  # 24 hours
 
-    def __init__(self, cron_class, *args, **kwargs):
+    cache: BaseCache
+    lock_name: str
+    timeout: float
+
+    def __init__(self, cron_class: type, *args, **kwargs):
         super().__init__(cron_class, *args, **kwargs)
 
         self.cache = self.get_cache_by_name()
@@ -57,12 +61,12 @@ def get_cache_by_name(self):
     def get_lock_name(self):
         return self.job_name
 
-    def get_cache_timeout(self, cron_class):
+    def get_cache_timeout(self, cron_class: type) -> float:
         try:
             timeout = getattr(
-                cron_class, 'DJANGO_CRON_LOCK_TIME', settings.DJANGO_CRON_LOCK_TIME
+                cron_class, "DJANGO_CRON_LOCK_TIME", settings.DJANGO_CRON_LOCK_TIME
             )
-        except:
+        except Exception:
             timeout = self.DEFAULT_LOCK_TIME
         return timeout
 

django_cron/backends/lock/database.py

@@ -1,7 +1,9 @@
-from django_cron.backends.lock.base import DjangoCronJobLock
-from django_cron.models import CronJobLock
 from django.db import transaction
 
+from django_cron.models import CronJobLock
+
+from .base import DjangoCronJobLock
+
 
 class DatabaseLock(DjangoCronJobLock):
     """
@@ -10,17 +12,17 @@ class DatabaseLock(DjangoCronJobLock):
     """
 
     @transaction.atomic
-    def lock(self):
-        lock, created = CronJobLock.objects.get_or_create(job_name=self.job_name)
+    def lock(self) -> bool:
+        lock, _ = CronJobLock.objects.get_or_create(job_name=self.job_name)
         if lock.locked:
             return False
-        else:
-            lock.locked = True
-            lock.save()
-            return True
+
+        lock.locked = True
+        lock.save(update_fields=["locked"])
+        return True
 
     @transaction.atomic
     def release(self):
-        lock = CronJobLock.objects.filter(job_name=self.job_name, locked=True).first()
+        lock = CronJobLock.objects.filter(job_name=self.job_name, locked=True)[0]
         lock.locked = False
-        lock.save()
+        lock.save(update_fields=["locked"])