diff --git a/django_q/admin.py b/django_q/admin.py index a61fd24..a1190cf 100644 --- a/django_q/admin.py +++ b/django_q/admin.py @@ -10,14 +10,7 @@ from django_q.tasks import async_task class TaskAdmin(admin.ModelAdmin): """model admin for success tasks.""" - list_display = ( - 'name', - 'func', - 'started', - 'stopped', - 'time_taken', - 'group' - ) + list_display = ("name", "func", "started", "stopped", "time_taken", "group") def has_add_permission(self, request): """Don't allow adds.""" @@ -28,9 +21,9 @@ class TaskAdmin(admin.ModelAdmin): qs = super(TaskAdmin, self).get_queryset(request) return qs.filter(success=True) - search_fields = ('name', 'func', 'group') + search_fields = ("name", "func", "group") readonly_fields = [] - list_filter = ('group',) + list_filter = ("group",) def get_readonly_fields(self, request, obj=None): """Set all fields readonly.""" @@ -50,21 +43,15 @@ retry_failed.short_description = _("Resubmit selected tasks to queue") class FailAdmin(admin.ModelAdmin): """model admin for failed tasks.""" - list_display = ( - 'name', - 'func', - 'started', - 'stopped', - 'short_result' - ) + list_display = ("name", "func", "started", "stopped", "short_result") def has_add_permission(self, request): """Don't allow adds.""" return False actions = [retry_failed] - search_fields = ('name', 'func') - list_filter = ('group',) + search_fields = ("name", "func") + list_filter = ("group",) readonly_fields = [] def get_readonly_fields(self, request, obj=None): @@ -76,31 +63,31 @@ class ScheduleAdmin(admin.ModelAdmin): """ model admin for schedules """ list_display = ( - 'id', - 'name', - 'func', - 'schedule_type', - 'repeats', - 'next_run', - 'last_run', - 'success' + "id", + "name", + "func", + "schedule_type", + "repeats", + "next_run", + "last_run", + "success", ) - list_filter = ('next_run', 'schedule_type') - search_fields = ('func',) - list_display_links = ('id', 'name') + # optional cron strings + try: + from croniter import croniter + except ImportError: + readonly_fields = ("cron",) + + list_filter = ("next_run", "schedule_type") + search_fields = ("func",) + list_display_links = ("id", "name") class QueueAdmin(admin.ModelAdmin): """ queue admin for ORM broker """ - list_display = ( - 'id', - 'key', - 'task_id', - 'name', - 'func', - 'lock' - ) + + list_display = ("id", "key", "task_id", "name", "func", "lock") def save_model(self, request, obj, form, change): obj.save(using=Conf.ORM) diff --git a/django_q/cluster.py b/django_q/cluster.py index 1b3c993..0ad51d3 100644 --- a/django_q/cluster.py +++ b/django_q/cluster.py @@ -1,6 +1,5 @@ -import ast - # Standard +import ast import importlib import signal import socket @@ -9,9 +8,8 @@ import uuid from multiprocessing import Event, Process, Value, current_process from time import sleep -# external +# External import arrow - # Django from django import db from django.conf import settings @@ -29,6 +27,12 @@ from django_q.signals import pre_execute from django_q.signing import SignedPackage, BadSignature from django_q.status import Stat, Status +# Optional +try: + from croniter import croniter +except ImportError: + croniter = None + class Cluster: def __init__(self, broker: Broker = None): @@ -103,10 +107,10 @@ class Cluster: @property def is_stopping(self) -> bool: return ( - self.stop_event - and self.start_event - and self.start_event.is_set() - and self.stop_event.is_set() + self.stop_event + and self.start_event + and self.start_event.is_set() + and self.stop_event.is_set() ) @property @@ -116,13 +120,13 @@ class Cluster: class Sentinel: def __init__( - self, - stop_event, - start_event, - cluster_id, - broker=None, - timeout=Conf.TIMEOUT, - start=True, + self, + stop_event, + start_event, + cluster_id, + broker=None, + timeout=Conf.TIMEOUT, + start=True, ): # Make sure we catch signals for the pool signal.signal(signal.SIGINT, signal.SIG_IGN) @@ -377,7 +381,7 @@ def monitor(result_queue: Queue, broker: Broker = None): def worker( - task_queue: Queue, result_queue: Queue, timer: Value, timeout: int = Conf.TIMEOUT + task_queue: Queue, result_queue: Queue, timer: Value, timeout: int = Conf.TIMEOUT ): """ Takes a task from the task queue, tries to execute it and puts the result back in the result queue @@ -552,9 +556,9 @@ def scheduler(broker: Broker = None): try: with db.transaction.atomic(using=Schedule.objects.db): for s in ( - Schedule.objects.select_for_update() - .exclude(repeats=0) - .filter(next_run__lt=timezone.now()) + Schedule.objects.select_for_update() + .exclude(repeats=0) + .filter(next_run__lt=timezone.now()) ): args = () kwargs = {} @@ -591,6 +595,16 @@ def scheduler(broker: Broker = None): next_run = next_run.shift(months=+3) elif s.schedule_type == s.YEARLY: next_run = next_run.shift(years=+1) + elif s.schedule_type == s.CRON: + if not croniter: + raise ImportError( + _( + "Please install croniter to enable cron expressions" + ) + ) + next_run = arrow.get( + croniter(s.cron, timezone.now()).get_next() + ) if Conf.CATCH_UP or next_run > arrow.utcnow(): break # arrow always returns a tz aware datetime, and we don't want diff --git a/django_q/models.py b/django_q/models.py index 86549ed..7eac054 100644 --- a/django_q/models.py +++ b/django_q/models.py @@ -1,16 +1,26 @@ +# Django from django import get_version +from django.core.exceptions import ValidationError +from django.db import models from django.template.defaultfilters import truncatechars - from django.urls import reverse +from django.utils import timezone from django.utils.html import format_html from django.utils.translation import gettext_lazy as _ -from django.db import models -from django.utils import timezone + +# External from picklefield import PickledObjectField from picklefield.fields import dbsafe_decode +# Local from django_q.signing import SignedPackage +# Optional +try: + from croniter import croniter +except ImportError: + croniter = None + class Task(models.Model): id = models.CharField(max_length=32, primary_key=True, editable=False) @@ -131,6 +141,18 @@ class Failure(Task): proxy = True +# Optional Cron validator +def validate_cron(value): + if not value: + return + if not croniter: + raise ImportError(_("Please install croniter to enable cron expressions")) + try: + croniter.expand(value) + except ValueError as e: + raise ValidationError(e) + + class Schedule(models.Model): name = models.CharField(max_length=100, null=True, blank=True) func = models.CharField(max_length=256, help_text="e.g. module.tasks.function") @@ -152,6 +174,7 @@ class Schedule(models.Model): MONTHLY = "M" QUARTERLY = "Q" YEARLY = "Y" + CRON = "C" TYPE = ( (ONCE, _("Once")), (MINUTES, _("Minutes")), @@ -161,6 +184,7 @@ class Schedule(models.Model): (MONTHLY, _("Monthly")), (QUARTERLY, _("Quarterly")), (YEARLY, _("Yearly")), + (CRON, _("Cron")), ) schedule_type = models.CharField( max_length=1, choices=TYPE, default=TYPE[0][0], verbose_name=_("Schedule Type") @@ -174,6 +198,13 @@ class Schedule(models.Model): next_run = models.DateTimeField( verbose_name=_("Next Run"), default=timezone.now, null=True ) + cron = models.CharField( + max_length=100, + null=True, + blank=True, + validators=[validate_cron], + help_text=_("Cron expression"), + ) task = models.CharField(max_length=100, null=True, editable=False) def success(self): diff --git a/poetry.lock b/poetry.lock index 28d26b9..55cee75 100644 --- a/poetry.lock +++ b/poetry.lock @@ -120,6 +120,18 @@ optional = false python-versions = ">=2.7, !=3.0.*, !=3.1.*, !=3.2.*, !=3.3.*, !=3.4.*" version = "0.4.3" +[[package]] +category = "main" +description = "croniter provides iteration for datetime object with cron like format" +name = "croniter" +optional = true +python-versions = ">=2.6, !=3.0.*, !=3.1.*, !=3.2.*, !=3.3.*" +version = "0.3.34" + +[package.dependencies] +natsort = "*" +python-dateutil = "*" + [[package]] category = "main" description = "A high-level Python Web framework that encourages rapid development and clean, pragmatic design." @@ -270,6 +282,18 @@ optional = false python-versions = ">=3.5" version = "8.4.0" +[[package]] +category = "main" +description = "Simple yet flexible natural sorting in Python." +name = "natsort" +optional = true +python-versions = ">=3.4" +version = "7.0.1" + +[package.extras] +fast = ["fastnumbers (>=2.0.0)"] +icu = ["PyICU (>=1.0.0)"] + [[package]] category = "dev" description = "Core utilities for Python packages" @@ -497,7 +521,7 @@ rollbar = ["django-q-rollbar"] sentry = [] [metadata] -content-hash = "d89c929e8b6951968228b1aab1aa5c0a7c014acceb3b986bbaa3e41df57d8b39" +content-hash = "6d89bff8bd465aa4a5facd812172a970bf5edebdd8099141054f72aa63260c2a" python-versions = ">=3.6" [metadata.files] @@ -545,6 +569,10 @@ colorama = [ {file = "colorama-0.4.3-py2.py3-none-any.whl", hash = "sha256:7d73d2a99753107a36ac6b455ee49046802e59d9d076ef8e47b61499fa29afff"}, {file = "colorama-0.4.3.tar.gz", hash = "sha256:e96da0d330793e2cb9485e9ddfd918d456036c7149416295932478192f4436a1"}, ] +croniter = [ + {file = "croniter-0.3.34-py2.py3-none-any.whl", hash = "sha256:15597ef0639f8fbab09cbf8c277fa8c65c8b9dbe818c4b2212f95dbc09c6f287"}, + {file = "croniter-0.3.34.tar.gz", hash = "sha256:7186b9b464f45cf3d3c83a18bc2344cc101d7b9fd35a05f2878437b14967e964"}, +] django = [ {file = "Django-3.0.7-py3-none-any.whl", hash = "sha256:e1630333248c9b3d4e38f02093a26f1e07b271ca896d73097457996e0fae12e8"}, {file = "Django-3.0.7.tar.gz", hash = "sha256:5052b34b34b3425233c682e0e11d658fd6efd587d11335a0203d827224ada8f2"}, @@ -638,6 +666,10 @@ more-itertools = [ {file = "more-itertools-8.4.0.tar.gz", hash = "sha256:68c70cc7167bdf5c7c9d8f6954a7837089c6a36bf565383919bb595efb8a17e5"}, {file = "more_itertools-8.4.0-py3-none-any.whl", hash = "sha256:b78134b2063dd214000685165d81c154522c3ee0a1c0d4d113c80361c234c5a2"}, ] +natsort = [ + {file = "natsort-7.0.1-py3-none-any.whl", hash = "sha256:d3fd728a3ceb7c78a59aa8539692a75e37cbfd9b261d4d702e8016639820f90a"}, + {file = "natsort-7.0.1.tar.gz", hash = "sha256:a633464dc3a22b305df0f27abcb3e83515898aa1fd0ed2f9726c3571a27258cf"}, +] packaging = [ {file = "packaging-20.4-py2.py3-none-any.whl", hash = "sha256:998416ba6962ae7fbd6596850b80e17859a5753ba17c32284f67bfff33784181"}, {file = "packaging-20.4.tar.gz", hash = "sha256:4357f74f47b9c12db93624a82154e9b120fa8293699949152b22065d556079f8"}, diff --git a/pyproject.toml b/pyproject.toml index 300f4c3..76ee572 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -1,6 +1,6 @@ [tool.poetry] name = "django-q" -version = "1.2.4" +version = "1.3.0" description = "A multiprocessing distributed task queue for Django" authors = ["Ilan Steemers "] license = "MIT" @@ -52,6 +52,7 @@ django-redis = {version = "^4.12.1", optional = true} iron-mq = {version = "^0.9", optional = true} boto3 = {version = "^1.14.12", optional = true} pymongo = {version = "^3.10.1", optional = true} +croniter = {version = "^0.3.34", optional = true} [tool.poetry.dev-dependencies] pytest = "^5.4.2" diff --git a/requirements.in b/requirements.in index 7f91d4c..fadec17 100644 --- a/requirements.in +++ b/requirements.in @@ -8,3 +8,4 @@ django-redis iron-mq boto3 pymongo +croniter diff --git a/requirements.txt b/requirements.txt index 1b3634d..464bd5e 100644 --- a/requirements.txt +++ b/requirements.txt @@ -11,6 +11,7 @@ boto3==1.14.12 # via -r requirements.in botocore==1.17.12 # via boto3, s3transfer certifi==2020.6.20 # via requests chardet==3.0.4 # via requests +croniter==0.3.34 # via -r requirements.in django-picklefield==3.0.1 # via -r requirements.in django-redis==4.12.1 # via -r requirements.in django==3.0.7 # via django-picklefield, django-redis @@ -20,9 +21,10 @@ idna==2.10 # via requests iron-core==1.2.0 # via iron-mq iron-mq==0.9 # via -r requirements.in jmespath==0.10.0 # via boto3, botocore +natsort==7.0.1 # via croniter psutil==5.7.0 # via -r requirements.in pymongo==3.10.1 # via -r requirements.in -python-dateutil==2.8.1 # via arrow, botocore, iron-core +python-dateutil==2.8.1 # via arrow, botocore, croniter, iron-core pytz==2020.1 # via django redis==3.5.3 # via -r requirements.in, django-redis requests==2.24.0 # via iron-core