delete old cron approach

Author: SKairinos

codeforlife/mixins/__init__.py

@@ -9,5 +9,4 @@
 from .allow_none import AllowNone
 from .base import BasePermission
 from .is_authenticated import IsAuthenticated
-from .is_cron_request_from_google import IsCronRequestFromGoogle
 from .operators import AND, NOT, OR, Permission

codeforlife/mixins/cron.py

@@ -37,4 +37,4 @@
 
 CELERY_BROKER_URL = REDIS_URL
 CELERY_TASK_TIME_LIMIT = 60 * 30
-CELERY_BEAT_SCHEDULE: t.Dict[str, t.Dict[str, CeleryBeat]] = {}
+CELERY_BEAT_SCHEDULE: t.Dict[str, CeleryBeat] = {}

codeforlife/permissions/__init__.py

@@ -7,37 +7,30 @@
 
 import typing as t
 
-from celery import schedules
 from celery import shared_task as _shared_task
+from celery.schedules import crontab, solar
 from django.conf import settings
 
 from .types import Args, KwArgs
 
+Schedule = t.Union[int, crontab, solar]
 
-class CeleryBeat(t.Dict[str, t.Any]):
+
+class CeleryBeat(t.TypedDict):
     """A Celery beat schedule.
 
     https://docs.celeryq.dev/en/v5.4.0/userguide/periodic-tasks.html
     """
 
-    # Shorthand for convenience.
-    crontab = schedules.crontab
-    solar = schedules.solar
-
-    def __init__(
-        self,
-        task: str,
-        schedule: t.Union[int, schedules.crontab, schedules.solar],
-        args: t.Optional[Args] = None,
-        kwargs: t.Optional[KwArgs] = None,
-    ):
-        super().__init__()
-        self["task"] = f"{settings.SERVICE_NAME}.{task}"
-        self["schedule"] = schedule
-        if args:
-            self["args"] = args
-        if kwargs:
-            self["kwargs"] = kwargs
+    task: str
+    schedule: Schedule
+    args: t.NotRequired[Args]
+    kwargs: t.NotRequired[KwArgs]
+
+    def __init__(self, **kwargs):  # type: ignore[misc]
+        kwargs["task"] = f"{settings.SERVICE_NAME}.{kwargs['task']}"
+
+        super().__init__(**kwargs)
 
 
 def shared_task(*args, **kwargs):

codeforlife/permissions/is_cron_request_from_google.py

@@ -8,7 +8,7 @@
 from .api import APITestCase, BaseAPITestCase
 from .api_client import APIClient, BaseAPIClient
 from .api_request_factory import APIRequestFactory, BaseAPIRequestFactory
-from .cron import CronTestCase
+from .celery import CeleryTestCase
 from .model import ModelTestCase
 from .model_list_serializer import (
     BaseModelListSerializerTestCase,

codeforlife/settings/third_party.py

@@ -0,0 +1,58 @@
+"""
+© Ocado Group
+Created on 01/04/2025 at 16:57:19(+01:00).
+"""
+
+from importlib import import_module
+from unittest import TestCase
+
+from celery import Celery, Task
+from django.conf import settings
+
+from ..tasks import CeleryBeat
+from ..types import Args, KwArgs
+
+
+class CeleryTestCase(TestCase):
+    """A test case for celery tasks."""
+
+    # The dot-path of the module containing the Celery app.
+    app_module: str = "application"
+    # The name of the Celery app.
+    app_name: str = "celery_app"
+    # The Celery app instance. Auto-imported if not set.
+    app: Celery
+
+    @classmethod
+    def setUpClass(cls):
+        if not hasattr(cls, "app"):
+            cls.app = getattr(import_module(cls.app_module), cls.app_name)
+
+    def apply_periodic_task(self, beat_name: str):
+        """Apply a periodic task.
+
+        Args:
+            beat_name: The name of the beat in the schedule.
+        """
+        beat: CeleryBeat = self.app.conf.beat_schedule[beat_name]
+
+        task_dot_path = (
+            beat["task"].removeprefix(f"{settings.SERVICE_NAME}.").split(".")
+        )
+        task_module = ".".join(task_dot_path[:-1])
+        task_name = task_dot_path[-1]
+        task: Task = getattr(import_module(task_module), task_name)
+
+        args, kwargs = beat.get("args", tuple()), beat.get("kwargs", {})
+        task.apply(*args, **kwargs)
+
+    def apply_task(self, name: str, args: Args, kwargs: KwArgs):
+        """Apply a task.
+
+        Args:
+            name: The name of the task.
+            args: The args to pass to the task.
+            kwargs: The keyword args to pass to the task.
+        """
+        task: Task = self.app.tasks[f"{settings.SERVICE_NAME}.{name}"]
+        task.apply(*args, **kwargs)

codeforlife/tasks.py

@@ -599,26 +599,6 @@ def bulk_destroy(
 
         return response
 
-    # --------------------------------------------------------------------------
-    # OTHER
-    # --------------------------------------------------------------------------
-
-    def cron_job(self, action: str):
-        """Call a CRON job.
-
-        Args:
-            action: The name of the action.
-
-        Returns:
-            The HTTP response.
-        """
-        response: Response = self.get(
-            self._test_case.reverse_action(action),
-            HTTP_X_APPENGINE_CRON="true",
-        )
-
-        return response
-
 
 # pylint: enable=no-member
 

codeforlife/tests/__init__.py

@@ -7,7 +7,7 @@
 from .api_root import APIRootView
 from .base_login import BaseLoginView
 from .csrf import CsrfCookieView
-from .decorators import action, cron_job
+from .decorators import action
 from .health_check import HealthCheckView
 from .model import BaseModelViewSet, ModelViewSet
 from .session import LogoutView, session_expired_view

codeforlife/tests/celery.py

@@ -52,19 +52,3 @@ def wrapper(handler: t.Callable):
         return _action(**kwargs)(handler)
 
     return wrapper
-
-
-def cron_job(handler: t.Callable):
-    # pylint: disable=line-too-long
-    """
-    Create an action that is meant to be called by Google as a CRON job.
-
-    See https://github.com/ocadotechnology/codeforlife-workspace/blob/main/cron.yaml.
-    """
-    # pylint: enable=line-too-long
-
-    return action(
-        detail=False,
-        methods=["get"],
-        url_path=f"cron/{_handler_name_to_url_path(handler)}",
-    )(handler)