refactor code

Author: cunla

docs/changelog.md

@@ -1,5 +1,11 @@
 # Changelog
 
+## v4.0.0b3 🌈
+
+Refactor the code to make it more organized and easier to maintain. This includes:
+- All types are under `types`.
+- 
+
 ## v4.0.0b2 🌈
 
 ### 🐛 Bug Fixes

docs/configuration.md

@@ -5,9 +5,9 @@
 All default settings for scheduler can be in one dictionary in `settings.py`:
 
 ```python
+import os
 from typing import Dict
-
-from scheduler.settings_types import SchedulerConfiguration, Broker, QueueConfiguration, UnixSignalDeathPenalty
+from scheduler.types import SchedulerConfiguration, Broker, QueueConfiguration
 
 
 SCHEDULER_CONFIG = SchedulerConfiguration(
@@ -25,7 +25,6 @@ SCHEDULER_CONFIG = SchedulerConfiguration(
     DEFAULT_MAINTENANCE_TASK_INTERVAL=10 * 60,  # The interval to run maintenance tasks in seconds. 10 minutes.
     DEFAULT_JOB_MONITORING_INTERVAL=30,  # The interval to monitor jobs in seconds.
     SCHEDULER_FALLBACK_PERIOD_SECS=120,  # Period (secs) to wait before requiring to reacquire locks
-    DEATH_PENALTY_CLASS=UnixSignalDeathPenalty,
 )
 SCHEDULER_QUEUES: Dict[str, QueueConfiguration] = {
     'default': QueueConfiguration(

docs/installation.md

@@ -19,8 +19,9 @@
    Note that the usage of `QueueConfiguration` is optional, you can use a simple dictionary, but `QueueConfiguration`
    helps preventing configuration errors.
    ```python
+    import os
     from typing import Dict
-    from scheduler.settings_types import QueueConfiguration
+    from scheduler.types import QueueConfiguration
 
     SCHEDULER_QUEUES: Dict[str, QueueConfiguration] = {
      'default': QueueConfiguration(
@@ -52,7 +53,7 @@
 
 4. Optional: Configure default values for queuing jobs from code:
    ```python
-   from scheduler.settings_types import SchedulerConfiguration, Broker,  UnixSignalDeathPenalty
+   from scheduler.types import SchedulerConfiguration, Broker
 
    SCHEDULER_CONFIG = SchedulerConfiguration(
     EXECUTIONS_IN_PAGE=20,
@@ -69,7 +70,6 @@
     DEFAULT_MAINTENANCE_TASK_INTERVAL=10 * 60,  # The interval to run maintenance tasks in seconds. 10 minutes.
     DEFAULT_JOB_MONITORING_INTERVAL=30,  # The interval to monitor jobs in seconds.
     SCHEDULER_FALLBACK_PERIOD_SECS=120,  # Period (secs) to wait before requiring to reacquire locks
-    DEATH_PENALTY_CLASS=UnixSignalDeathPenalty,
    )
    ```
 

scheduler/__init__.py

@@ -3,10 +3,7 @@
 __version__ = importlib.metadata.version("django-tasks-scheduler")
 
 __all__ = [
-    "QueueConfiguration",
-    "SchedulerConfiguration",
     "job",
 ]
 
-from scheduler.settings_types import QueueConfiguration, SchedulerConfiguration
 from scheduler.decorators import job

scheduler/admin/ephemeral_models.py

@@ -1,8 +1,7 @@
 from django.contrib import admin
 
 from scheduler import views
-from scheduler.models.queue import Queue
-from scheduler.models.worker import Worker
+from scheduler.models.ephemeral_models import Queue, Worker
 
 
 class ImmutableAdmin(admin.ModelAdmin):

scheduler/admin/task_admin.py

@@ -6,13 +6,11 @@
 from django.http import HttpRequest
 from django.utils.translation import gettext_lazy as _
 
-from scheduler.broker_types import ConnectionErrorTypes
-from scheduler.helpers import tools
 from scheduler.helpers.queues import get_queue
-from scheduler.models.args import TaskArg, TaskKwarg
-from scheduler.models.task import Task, TaskType
+from scheduler.models import TaskArg, TaskKwarg, Task, TaskType, get_next_cron_time
 from scheduler.redis_models import JobModel
 from scheduler.settings import SCHEDULER_CONFIG, logger
+from scheduler.types import ConnectionErrorTypes
 
 
 def job_execution_of(job: JobModel, task: Task) -> bool:
@@ -142,7 +140,7 @@ def task_schedule(self, o: Task) -> str:
 
     @admin.display(description="Next run")
     def next_run(self, o: Task) -> str:
-        return tools.get_next_cron_time(o.cron_string)
+        return get_next_cron_time(o.cron_string)
 
     def change_view(self, request: HttpRequest, object_id, form_url="", extra_context=None):
         extra = extra_context or {}

scheduler/decorators.py

@@ -1,7 +1,7 @@
 from functools import wraps
 from typing import Any, Callable, Dict, Optional, Union
 
-from scheduler.broker_types import ConnectionType
+from scheduler.types import ConnectionType
 from scheduler.helpers.callback import Callback
 
 JOB_METHODS_LIST = list()

scheduler/helpers/queues/getters.py

@@ -1,6 +1,6 @@
 from typing import List, Set
 
-from scheduler.broker_types import ConnectionErrorTypes, BrokerMetaData
+from scheduler.types import ConnectionErrorTypes, BrokerMetaData
 from scheduler.redis_models.worker import WorkerModel
 from scheduler.settings import (
     SCHEDULER_CONFIG,

scheduler/helpers/queues/queue_logic.py

@@ -7,7 +7,6 @@
 
 from redis import WatchError
 
-from scheduler.broker_types import ConnectionType, FunctionReferenceType
 from scheduler.helpers.callback import Callback
 from scheduler.helpers.utils import utcnow, current_timestamp
 from scheduler.redis_models import (
@@ -22,6 +21,7 @@
 )
 from scheduler.redis_models import JobStatus, SchedulerLock, Result, ResultType, JobModel
 from scheduler.settings import logger, SCHEDULER_CONFIG
+from scheduler.types import ConnectionType, FunctionReferenceType
 
 
 class InvalidJobOperation(Exception):
@@ -100,6 +100,7 @@ def clean_registries(self, timestamp: Optional[float] = None) -> None:
         Removed jobs are added to the global failed job queue.
         """
         before_score = timestamp or current_timestamp()
+        self.queued_job_registry.compact()
         started_jobs: List[Tuple[str, float]] = self.active_job_registry.get_job_names_before(
             self.connection, before_score
         )
@@ -119,15 +120,12 @@ def clean_registries(self, timestamp: Optional[float] = None) -> None:
 
                 else:
                     logger.warning(
-                        f"{self.__class__.__name__} cleanup: Moving job to {self.failed_job_registry.key} "
-                        f"(due to AbandonedJobError)"
+                        f"Queue cleanup: Moving job to {self.failed_job_registry.key} (due to AbandonedJobError)"
                     )
-                    job.set_status(JobStatus.FAILED, connection=pipeline)
                     exc_string = (
                         f"Moved to {self.failed_job_registry.key}, due to AbandonedJobError, at {datetime.now()}"
                     )
-                    job.save(connection=pipeline)
-                    job.expire(ttl=-1, connection=pipeline)
+                    job.status = JobStatus.FAILED
                     score = current_timestamp() + SCHEDULER_CONFIG.DEFAULT_FAILURE_TTL
                     Result.create(
                         connection=pipeline,
@@ -138,8 +136,8 @@ def clean_registries(self, timestamp: Optional[float] = None) -> None:
                         exc_string=exc_string,
                     )
                     self.failed_job_registry.add(pipeline, job.name, score)
-                    job.save(connection=pipeline)
                     job.expire(connection=pipeline, ttl=SCHEDULER_CONFIG.DEFAULT_FAILURE_TTL)
+                    job.save(connection=pipeline)
 
             for registry in self.REGISTRIES.values():
                 getattr(self, registry).cleanup(connection=self.connection, timestamp=before_score)
@@ -188,24 +186,24 @@ def get_all_jobs(self) -> List[JobModel]:
         return JobModel.get_many(job_names, connection=self.connection)
 
     def create_and_enqueue_job(
-        self,
-        func: FunctionReferenceType,
-        args: Union[Tuple, List, None] = None,
-        kwargs: Optional[Dict] = None,
-        timeout: Optional[int] = None,
-        result_ttl: Optional[int] = None,
-        job_info_ttl: Optional[int] = None,
-        description: Optional[str] = None,
-        name: Optional[str] = None,
-        at_front: bool = False,
-        meta: Optional[Dict] = None,
-        on_success: Optional[Callback] = None,
-        on_failure: Optional[Callback] = None,
-        on_stopped: Optional[Callback] = None,
-        task_type: Optional[str] = None,
-        scheduled_task_id: Optional[int] = None,
-        when: Optional[datetime] = None,
-        pipeline: Optional[ConnectionType] = None,
+            self,
+            func: FunctionReferenceType,
+            args: Union[Tuple, List, None] = None,
+            kwargs: Optional[Dict] = None,
+            timeout: Optional[int] = None,
+            result_ttl: Optional[int] = None,
+            job_info_ttl: Optional[int] = None,
+            description: Optional[str] = None,
+            name: Optional[str] = None,
+            at_front: bool = False,
+            meta: Optional[Dict] = None,
+            on_success: Optional[Callback] = None,
+            on_failure: Optional[Callback] = None,
+            on_stopped: Optional[Callback] = None,
+            task_type: Optional[str] = None,
+            scheduled_task_id: Optional[int] = None,
+            when: Optional[datetime] = None,
+            pipeline: Optional[ConnectionType] = None,
     ) -> JobModel:
         """Creates a job to represent the delayed function call and enqueues it.
         :param when: When to schedule the job (None to enqueue immediately)
@@ -312,7 +310,7 @@ def run_job(self, job: JobModel) -> JobModel:
         return job
 
     def enqueue_job(
-        self, job_model: JobModel, connection: Optional[ConnectionType] = None, at_front: bool = False
+            self, job_model: JobModel, connection: Optional[ConnectionType] = None, at_front: bool = False
     ) -> JobModel:
         """Enqueues a job for delayed execution without checking dependencies.
 
@@ -363,10 +361,10 @@ def run_sync(self, job: JobModel) -> JobModel:
 
     @classmethod
     def dequeue_any(
-        cls,
-        queues: List[Self],
-        timeout: Optional[int],
-        connection: Optional[ConnectionType] = None,
+            cls,
+            queues: List[Self],
+            timeout: Optional[int],
+            connection: Optional[ConnectionType] = None,
     ) -> Tuple[Optional[JobModel], Optional[Self]]:
         """Class method returning a Job instance at the front of the given set of Queues, where the order of the queues
         is important.
@@ -382,6 +380,8 @@ def dequeue_any(
 
         while True:
             registries = [q.queued_job_registry for q in queues]
+            for registry in registries:
+                registry.compact()
 
             registry_key, job_name = QueuedJobRegistry.pop(connection, registries, timeout)
             if job_name is None: