wip

Author: cunla

scheduler/management/commands/export.py

@@ -54,7 +54,7 @@ def handle(self, *args, **options):
         if options.get("format") == "json":
             import json
 
-            click.echo(json.dumps(res, indent=2), file=file)
+            click.echo(json.dumps(res, indent=2, default=str), file=file)
             return
 
         if options.get("format") == "yaml":

scheduler/management/commands/import.py

@@ -8,7 +8,7 @@
 from django.core.management.base import BaseCommand
 from django.utils import timezone
 
-from scheduler.models import TaskArg, TaskKwarg
+from scheduler.models import TaskArg, TaskKwarg, Task
 from scheduler.tools import MODEL_NAMES
 
 
@@ -17,18 +17,29 @@ def job_model_str(model_str: str) -> str:
         return model_str[:-3] + "Task"
     return model_str
 
-
-def create_job_from_dict(job_dict: Dict[str, Any], update):
-    model = apps.get_model(app_label="scheduler", model_name=job_model_str(job_dict["model"]))
-    existing_job = model.objects.filter(name=job_dict["name"]).first()
+def get_task_type(model_str: str) -> Task.TaskType:
+    model_str = job_model_str(model_str)
+    if model_str not in MODEL_NAMES:
+        raise ValueError(f"Invalid model {model_str}")
+    if model_str == "CronTask":
+        return Task.TaskType.CRON
+    elif model_str == "RepeatableTask":
+        return Task.TaskType.REPEATABLE
+    elif model_str == "ScheduledTask":
+        return Task.TaskType.ONCE
+
+def create_task_from_dict(task_dict: Dict[str, Any], update):
+    existing_job = Task.objects.filter(name=task_dict["name"]).first()
+    task_type = get_task_type(task_dict["model"])
     if existing_job:
         if update:
             click.echo(f'Found existing job "{existing_job}, removing it to be reinserted"')
             existing_job.delete()
         else:
             click.echo(f'Found existing job "{existing_job}", skipping')
             return
-    kwargs = dict(job_dict)
+    kwargs = dict(task_dict)
+    kwargs["task_type"] = task_type
     del kwargs["model"]
     del kwargs["callable_args"]
     del kwargs["callable_kwargs"]
@@ -37,21 +48,21 @@ def create_job_from_dict(job_dict: Dict[str, Any], update):
         if not settings.USE_TZ and not timezone.is_naive(target):
             target = timezone.make_naive(target)
         kwargs["scheduled_time"] = target
-    model_fields = set(map(lambda field: field.attname, model._meta.get_fields()))
+    model_fields = set(map(lambda field: field.attname, Task._meta.get_fields()))
     keys_to_ignore = list(filter(lambda _k: _k not in model_fields, kwargs.keys()))
     for k in keys_to_ignore:
         del kwargs[k]
-    scheduled_job = model.objects.create(**kwargs)
+    scheduled_job = Task.objects.create(**kwargs)
     click.echo(f"Created job {scheduled_job}")
     content_type = ContentType.objects.get_for_model(scheduled_job)
 
-    for arg in job_dict["callable_args"]:
+    for arg in task_dict["callable_args"]:
         TaskArg.objects.create(
             content_type=content_type,
             object_id=scheduled_job.id,
             **arg,
         )
-    for arg in job_dict["callable_kwargs"]:
+    for arg in task_dict["callable_kwargs"]:
         TaskKwarg.objects.create(
             content_type=content_type,
             object_id=scheduled_job.id,
@@ -125,4 +136,4 @@ def handle(self, *args, **options):
                 model.objects.all().delete()
 
         for job in jobs:
-            create_job_from_dict(job, update=options.get("update"))
+            create_task_from_dict(job, update=options.get("update"))

scheduler/rq_classes.py

@@ -61,9 +61,9 @@ def __eq__(self, other) -> bool:
     def is_scheduled_task(self) -> bool:
         return self.meta.get("scheduled_task_id", None) is not None
 
-    def is_execution_of(self, task: "ScheduledTask") -> bool:  # noqa: F821
+    def is_execution_of(self, task: "Task") -> bool:  # noqa: F821
         return (
-            self.meta.get("task_type", None) == task.TASK_TYPE and self.meta.get("scheduled_task_id", None) == task.id
+            self.meta.get("task_type", None) == task.task_type and self.meta.get("scheduled_task_id", None) == task.id
         )
 
     def stop_execution(self, connection: ConnectionType):

scheduler/tools.py

@@ -1,6 +1,6 @@
 import importlib
 import os
-from typing import List, Any, Callable
+from typing import List, Any, Callable, Optional
 
 import croniter
 from django.apps import apps
@@ -21,8 +21,10 @@ def callable_func(callable_str: str) -> Callable:
     return func
 
 
-def get_next_cron_time(cron_string) -> timezone.datetime:
+def get_next_cron_time(cron_string: Optional[str]) -> Optional[timezone.datetime]:
     """Calculate the next scheduled time by creating a crontab object with a cron string"""
+    if cron_string is None:
+        return None
     now = timezone.now()
     itr = croniter.croniter(cron_string, now)
     next_itr = itr.get_next(timezone.datetime)

testproject/testproject/settings.py

@@ -44,9 +44,9 @@
     "default": {
         "BACKEND": "django.core.cache.backends.redis.RedisCache",
         "LOCATION": [
-            "redis://127.0.0.1:6379",  # leader
+            "redis://127.0.0.1:6379",
         ],
-        "OPTIONS": {"connection_class": FakeConnection},
+        "BROKER": "fakeredis",
     }
 }
 TEMPLATES = [
@@ -115,19 +115,19 @@
 STATIC_URL = "/static/"
 SCHEDULER_QUEUES = {
     "default": {
-        "URL": f"redis://localhost:${BROKER_PORT}/0",
+        "URL": f"redis://localhost:{BROKER_PORT}/0",
     },
     "low": {
-        "URL": f"redis://localhost:${BROKER_PORT}/0",
+        "URL": f"redis://localhost:{BROKER_PORT}/0",
     },
     "high": {
-        "URL": f"redis://localhost:${BROKER_PORT}/1",
+        "URL": f"redis://localhost:{BROKER_PORT}/1",
     },
     "medium": {
-        "URL": f"redis://localhost:${BROKER_PORT}/1",
+        "URL": f"redis://localhost:{BROKER_PORT}/1",
     },
     "another": {
-        "URL": f"redis://localhost:${BROKER_PORT}/1",
+        "URL": f"redis://localhost:{BROKER_PORT}/1",
     },
 }
 DEFAULT_AUTO_FIELD = "django.db.models.BigAutoField"