Merge branch 'master' into save_safety

Author: cunla

docs/changelog.md

@@ -1,5 +1,12 @@
 # Changelog
 
+## v4.0.8 🌈
+
+### 🐛 Bug Fixes
+
+- Fix: Scheduled jobs fail to execute: {'scheduled_time': ['Scheduled time must be in the future']} #297
+- Fix Error with deserialize JobModel due to multi-processing #291
+
 ## v4.0.7 🌈
 
 ### 🧰 Maintenance

docs/requirements.txt

@@ -1,2 +1,2 @@
 mkdocs==1.6.1
-mkdocs-material==9.6.19
+mkdocs-material==9.6.20

pyproject.toml

@@ -4,7 +4,7 @@ build-backend = "hatchling.build"
 
 [project]
 name = "django-tasks-scheduler"
-version = "4.0.7"
+version = "4.0.8"
 description = "An async job scheduler for django using redis/valkey brokers"
 authors = [{ name = "Daniel Moran", email = "[email protected]" }]
 requires-python = ">=3.10"
@@ -56,12 +56,12 @@ Funding = "https://github.com/sponsors/cunla"
 
 [dependency-groups]
 dev = [
-    "time-machine>=2.16.0",
-    "ruff>=0.11",
-    "coverage>=7.6",
+    "time-machine>=2.19",
+    "ruff>=0.13",
+    "coverage[toml]>=7.10",
     "fakeredis>=2.28",
-    "pyyaml>=6,<7",
-    "mypy>=1.16.0",
+    "pyyaml>=6",
+    "mypy>=1.18",
     "types-croniter>=6.0.0.20250411",
     "beautifulsoup4>=4.13.4"
 ]
@@ -92,9 +92,9 @@ line-ending = "auto"
 [tool.mypy]
 packages = ['scheduler', ]
 exclude = ["scheduler/tests/.*\\.py",
-           "scheduler/migrations/.*\\.py",
-           "testproject/.*\\.py",
-           "testproject/tests/.*\\.py"]
+    "scheduler/migrations/.*\\.py",
+    "testproject/.*\\.py",
+    "testproject/tests/.*\\.py"]
 strict = true
 follow_imports = "silent"
 ignore_missing_imports = true

scheduler/helpers/queues/getters.py

@@ -46,6 +46,13 @@ def _get_connection(config: QueueConfiguration, use_strict_broker: bool = False)
     )
 
 
+def refresh_queue_connection(queue: Queue) -> None:
+    """Refreshes the connection of a given Queue"""
+    queue_settings = get_queue_configuration(queue.name)
+    connection = _get_connection(queue_settings)
+    queue.refresh_connection(connection)
+
+
 def get_queue(name: str = "default") -> Queue:
     """Returns an DjangoQueue using parameters defined in `SCHEDULER_QUEUES`"""
     queue_settings = get_queue_configuration(name)

scheduler/helpers/queues/queue_logic.py

@@ -44,17 +44,16 @@ def perform_job(job_model: JobModel, connection: ConnectionType) -> Any:  # noqa
 
     try:
         result = job_model.func(*job_model.args, **job_model.kwargs)
+        job_model.save(connection=connection, save_all=True)
         if asyncio.iscoroutine(result):
             loop = asyncio.new_event_loop()
             coro_result = loop.run_until_complete(result)
             result = coro_result
-        if job_model.success_callback:
-            job_model.success_callback(job_model, connection, result)
+        job_model.call_success_callback(job_model, connection, result)
         return result
     except Exception as e:
         logger.error(f"Job {job_model.name} failed with exception: {e}", exc_info=True)
-        if job_model.failure_callback:
-            job_model.failure_callback(job_model, connection, *sys.exc_info())
+        job_model.call_failure_callback(job_model, connection, *sys.exc_info())
         raise
     finally:
         assert job_model is _job_stack.pop()
@@ -90,6 +89,15 @@ def __init__(self, connection: ConnectionType, name: str, is_async: bool = True)
         self.scheduled_job_registry = ScheduledJobRegistry(connection=self.connection, name=self.name)
         self.canceled_job_registry = CanceledJobRegistry(connection=self.connection, name=self.name)
 
+    def refresh_connection(self, connection: ConnectionType) -> None:
+        self.connection = connection
+        self.queued_job_registry.connection = connection
+        self.active_job_registry.connection = connection
+        self.failed_job_registry.connection = connection
+        self.finished_job_registry.connection = connection
+        self.scheduled_job_registry.connection = connection
+        self.canceled_job_registry.connection = connection
+
     def __len__(self) -> int:
         return self.count
 
@@ -111,43 +119,27 @@ def clean_registries(self, timestamp: Optional[float] = None) -> None:
             self.connection, before_score
         )
 
-        with self.connection.pipeline() as pipeline:
-            for job_name, job_score in started_jobs:
-                job = JobModel.get(job_name, connection=self.connection)
-                if job is None or job.failure_callback is None or job_score + job.timeout > before_score:
-                    continue
+        for job_name, job_score in started_jobs:
+            job = JobModel.get(job_name, connection=self.connection)
+            if job is None or not job.has_failure_callback or job_score + job.timeout > before_score:
+                continue
 
-                logger.debug(f"Running failure callbacks for {job.name}")
-                try:
-                    job.failure_callback(job, self.connection, traceback.extract_stack())
-                except Exception:  # noqa
-                    logger.exception(f"Job {self.name}: error while executing failure callback")
-                    raise
+            logger.debug(f"Running failure callbacks for {job.name}")
+            try:
+                job.call_failure_callback(job, self.connection, traceback.extract_stack())
+            except Exception:  # noqa
+                logger.exception(f"Job {self.name}: error while executing failure callback")
+                raise
 
-                else:
-                    logger.warning(
-                        f"Queue cleanup: Moving job to {self.failed_job_registry.key} (due to AbandonedJobError)"
-                    )
-                    exc_string = (
-                        f"Moved to {self.failed_job_registry.key}, due to AbandonedJobError, at {datetime.now()}"
-                    )
-                    job.status = JobStatus.FAILED
-                    score = current_timestamp() + SCHEDULER_CONFIG.DEFAULT_FAILURE_TTL
-                    Result.create(
-                        connection=pipeline,
-                        job_name=job.name,
-                        worker_name=job.worker_name,
-                        _type=ResultType.FAILED,
-                        ttl=SCHEDULER_CONFIG.DEFAULT_FAILURE_TTL,
-                        exc_string=exc_string,
-                    )
-                    self.failed_job_registry.add(pipeline, job.name, score)
-                    job.expire(connection=pipeline, ttl=SCHEDULER_CONFIG.DEFAULT_FAILURE_TTL)
-                    job.save(connection=pipeline)
+            else:
+                logger.warning(
+                    f"Queue cleanup: Moving job to {self.failed_job_registry.key} (due to AbandonedJobError)"
+                )
+                exc_string = f"Moved to {self.failed_job_registry.key}, due to AbandonedJobError, at {datetime.now()}"
+                self.job_handle_failure(JobStatus.FAILED, job, exc_string)
 
             for registry in self.REGISTRIES.values():
                 getattr(self, registry).cleanup(connection=self.connection, timestamp=before_score)
-            pipeline.execute()
 
     def first_queued_job_name(self) -> Optional[str]:
         return self.queued_job_registry.get_first()
@@ -248,37 +240,35 @@ def create_and_enqueue_job(
             raise TypeError(f"Invalid type for when=`{when}`")
         return job_model
 
-    def job_handle_success(
-        self, job: JobModel, result: Any, job_info_ttl: int, result_ttl: int, connection: ConnectionType
-    ) -> None:
+    def job_handle_success(self, job: JobModel, result: Any, job_info_ttl: int, result_ttl: int) -> None:
         """Saves and cleanup job after successful execution"""
         job.after_execution(
             job_info_ttl,
             JobStatus.FINISHED,
             prev_registry=self.active_job_registry,
             new_registry=self.finished_job_registry,
-            connection=connection,
+            connection=self.connection,
         )
         Result.create(
-            connection,
+            self.connection,
             job_name=job.name,
             worker_name=job.worker_name,
             _type=ResultType.SUCCESSFUL,
             return_value=result,
             ttl=result_ttl,
         )
 
-    def job_handle_failure(self, status: JobStatus, job: JobModel, exc_string: str, connection: ConnectionType) -> None:
+    def job_handle_failure(self, status: JobStatus, job: JobModel, exc_string: str) -> None:
         # Does not set job status since the job might be stopped
         job.after_execution(
             SCHEDULER_CONFIG.DEFAULT_FAILURE_TTL,
             status,
             prev_registry=self.active_job_registry,
             new_registry=self.failed_job_registry,
-            connection=connection,
+            connection=self.connection,
         )
         Result.create(
-            connection,
+            self.connection,
             job.name,
             job.worker_name,
             ResultType.FAILED,
@@ -291,19 +281,11 @@ def run_sync(self, job: JobModel) -> JobModel:
         job.prepare_for_execution("sync", self.active_job_registry, self.connection)
         try:
             result = perform_job(job, self.connection)
-
-            with self.connection.pipeline() as pipeline:
-                self.job_handle_success(
-                    job, result=result, job_info_ttl=job.job_info_ttl, result_ttl=job.success_ttl, connection=pipeline
-                )
-
-                pipeline.execute()
+            self.job_handle_success(job, result=result, job_info_ttl=job.job_info_ttl, result_ttl=job.success_ttl)
         except Exception as e:  # noqa
             logger.warning(f"Job {job.name} failed with exception: {e}")
-            with self.connection.pipeline() as pipeline:
-                exc_string = "".join(traceback.format_exception(*sys.exc_info()))
-                self.job_handle_failure(JobStatus.FAILED, job, exc_string, pipeline)
-                pipeline.execute()
+            exc_string = "".join(traceback.format_exception(*sys.exc_info()))
+            self.job_handle_failure(JobStatus.FAILED, job, exc_string)
         return job
 
     @classmethod

scheduler/models/task.py

@@ -289,20 +289,9 @@ def to_dict(self) -> Dict[str, Any]:
             model=str(self.task_type),
             name=self.name,
             callable=self.callable,
-            callable_args=[
-                dict(
-                    arg_type=arg.arg_type,
-                    val=arg.val,
-                )
-                for arg in self.callable_args.all()
-            ],
+            callable_args=[dict(arg_type=arg.arg_type, val=arg.val) for arg in self.callable_args.all()],
             callable_kwargs=[
-                dict(
-                    arg_type=arg.arg_type,
-                    key=arg.key,
-                    val=arg.val,
-                )
-                for arg in self.callable_kwargs.all()
+                dict(arg_type=arg.arg_type, key=arg.key, val=arg.val) for arg in self.callable_kwargs.all()
             ],
             enabled=self.enabled,
             queue=self.queue,
@@ -344,23 +333,18 @@ def _schedule(self) -> bool:
             logger.debug(f"Task {str(self)} scheduled time is in the past, not scheduling")
             return False
         kwargs = self._enqueue_args()
-        job = self.rqueue.create_and_enqueue_job(
-            run_task,
-            args=(self.task_type, self.id),
-            when=schedule_time,
-            **kwargs,
-        )
+        job = self.rqueue.create_and_enqueue_job(run_task, args=(self.task_type, self.id), when=schedule_time, **kwargs)
         self.job_name = job.name
         return True
 
     def save(self, **kwargs: Any) -> None:
         should_clean = kwargs.pop("clean", True)
+        schedule_job = kwargs.pop("schedule_job", True)
         if should_clean:
             self.clean()
         update_fields = kwargs.get("update_fields", None)
         if update_fields is not None:
             kwargs["update_fields"] = set(update_fields).union({"updated_at"})
-        schedule_job = kwargs.pop("schedule_job", True)
         super(Task, self).save(**kwargs)
         if schedule_job:
             self._schedule()

scheduler/redis_models/base.py

@@ -108,7 +108,8 @@ def deserialize(cls, data: Dict[str, Any]) -> Self:
                 logger.warning(f"Unknown field {k} in {cls.__name__}")
                 continue
             data[k] = _deserialize(data[k], types[k])
-        return cls(**data)
+        res = cls(**data)
+        return res
 
 
 @dataclasses.dataclass(slots=True, kw_only=True)
@@ -118,6 +119,22 @@ class HashModel(BaseModel):
     _list_key: ClassVar[str] = ":list_all:"
     _children_key_template: ClassVar[str] = ":children:{}:"
 
+    def __post_init__(self):
+        self._dirty_fields = set()
+        self._save_all = True
+
+    def __setattr__(self, key, value):
+        if not key.startswith("_") and hasattr(self, "_dirty_fields"):
+            self._dirty_fields.add(key)
+        super(HashModel, self).__setattr__(key, value)
+
+    @classmethod
+    def deserialize(cls, data: Dict[str, Any]) -> Self:
+        instance = super(HashModel, cls).deserialize(data)
+        instance._dirty_fields = set()
+        instance._save_all = False
+        return instance
+
     @property
     def _parent_key(self) -> Optional[str]:
         if self.parent is None:
@@ -168,29 +185,33 @@ def get_many(cls, names: Sequence[str], connection: ConnectionType) -> List[Opti
             values = pipeline.execute()
             return [(cls.deserialize(decode_dict(v, set())) if v else None) for v in values]
 
-    def save(self, connection: ConnectionType) -> None:
+    def save(self, connection: ConnectionType, save_all: bool = False) -> None:
+        save_all = save_all or self._save_all
         with connection.pipeline() as pipeline:
             pipeline.sadd(self._list_key, self.name)
             if self._parent_key is not None:
                 pipeline.sadd(self._parent_key, self.name)
             mapping = self.serialize(with_nones=True)
+            if not save_all:
+                mapping = {k: v for k, v in mapping.items() if k in self._dirty_fields}
             none_values = {k for k, v in mapping.items() if v is None}
             if none_values:
                 pipeline.hdel(self._key, *none_values)
             mapping = {k: v for k, v in mapping.items() if v is not None}
             if mapping:
                 pipeline.hset(self._key, mapping=mapping)
-
             pipeline.execute()
+            self._dirty_fields = set()
+            self._save_all = False
 
     def delete(self, connection: ConnectionType) -> None:
         with connection.pipeline() as pipeline:
             pipeline.srem(self._list_key, self._key)
             if self._parent_key is not None:
                 pipeline.srem(self._parent_key, 0, self._key)
             pipeline.delete(self._key)
-
             pipeline.execute()
+            self._save_all = True
 
     @classmethod
     def count(cls, connection: ConnectionType, parent: Optional[str] = None) -> int: