black

Author: cunla

scheduler/admin/__init__.py

@@ -1,3 +1,3 @@
 from .task_models import TaskAdmin as OldTaskAdmin  # noqa: F401
 from .ephemeral_models import QueueAdmin, WorkerAdmin  # noqa: F401
-from .task_admin import TaskAdmin  # noqa: F401
+from .task_admin import TaskAdmin  # noqa: F401

scheduler/management/commands/rqworker.py

@@ -118,10 +118,7 @@ def handle(self, **options):
 
         try:
             # Instantiate a worker
-            w = create_worker(
-                *queues,
-                **init_options
-            )
+            w = create_worker(*queues, **init_options)
 
             # Close any opened DB connection before any fork
             reset_db_connections()

scheduler/models/scheduled_task.py

@@ -395,7 +395,7 @@ class ScheduledTask(ScheduledTimeMixin, BaseTask):
 
     def ready_for_schedule(self) -> bool:
         return super(ScheduledTask, self).ready_for_schedule() and (
-                self.scheduled_time is None or self.scheduled_time >= timezone.now()
+            self.scheduled_time is None or self.scheduled_time >= timezone.now()
         )
 
     class Meta:

scheduler/models/task.py

@@ -93,8 +93,7 @@ class TimeUnits(models.TextChoices):
     )
     queue = models.CharField(_("queue"), max_length=255, choices=get_queue_choices, help_text=_("Queue name"))
     job_id = models.CharField(
-        _("job id"), max_length=128, editable=False, blank=True, null=True,
-        help_text=_("Current job_id on queue")
+        _("job id"), max_length=128, editable=False, blank=True, null=True, help_text=_("Current job_id on queue")
     )
     at_front = models.BooleanField(
         _("At front"),
@@ -147,10 +146,18 @@ class TimeUnits(models.TextChoices):
         help_text=_("Last time the task has failed"),
     )
     interval = models.PositiveIntegerField(
-        _("interval"), blank=True, null=True,
-        help_text=_("Interval for repeatable task"), )
+        _("interval"),
+        blank=True,
+        null=True,
+        help_text=_("Interval for repeatable task"),
+    )
     interval_unit = models.CharField(
-        _("interval unit"), max_length=12, choices=TimeUnits.choices, default=TimeUnits.HOURS, blank=True, null=True,
+        _("interval unit"),
+        max_length=12,
+        choices=TimeUnits.choices,
+        default=TimeUnits.HOURS,
+        blank=True,
+        null=True,
     )
     repeat = models.PositiveIntegerField(
         _("repeat"),
@@ -161,7 +168,9 @@ class TimeUnits(models.TextChoices):
     scheduled_time = models.DateTimeField(_("scheduled time"))
     cron_string = models.CharField(
         _("cron string"),
-        max_length=64, blank=True, null=True,
+        max_length=64,
+        blank=True,
+        null=True,
         help_text=mark_safe(
             """Define the schedule in a crontab like syntax.
             Times are in UTC. Use <a href="https://crontab.guru/">crontab.guru</a> to create a cron string."""

scheduler/queues.py

@@ -87,7 +87,7 @@ def get_connection(queue_settings, use_strict_redis=False):
 
 
 def get_queue(
-        name="default", default_timeout=None, is_async=None, autocommit=None, connection=None, **kwargs
+    name="default", default_timeout=None, is_async=None, autocommit=None, connection=None, **kwargs
 ) -> DjangoQueue:
     """Returns an DjangoQueue using parameters defined in `SCHEDULER_QUEUES`"""
     from .settings import QUEUES

scheduler/rq_classes.py

@@ -62,8 +62,9 @@ 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
-        return (self.meta.get("task_type", None) == task.TASK_TYPE
-                and self.meta.get("scheduled_task_id", None) == task.id)
+        return (
+            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):
         send_stop_job_command(connection, self.id)
@@ -91,11 +92,11 @@ def __str__(self):
         return f"{self.name}/{','.join(self.queue_names())}"
 
     def _start_scheduler(
-            self,
-            burst: bool = False,
-            logging_level: str = "INFO",
-            date_format: str = "%H:%M:%S",
-            log_format: str = "%(asctime)s %(message)s",
+        self,
+        burst: bool = False,
+        logging_level: str = "INFO",
+        date_format: str = "%H:%M:%S",
+        log_format: str = "%(asctime)s %(message)s",
     ) -> None:
         """Starts the scheduler process.
         This is specifically designed to be run by the worker when running the `work()` method.