CeleryBeatScheduleBuilder

Author: SKairinos

codeforlife/tasks.py

@@ -31,13 +31,30 @@ class CeleryBeat(t.TypedDict):
 CeleryBeatSchedule = t.Dict[str, CeleryBeat]
 
 
-def celery_beat_schedule(**beat_schedule: CeleryBeat) -> CeleryBeatSchedule:
-    """Create a beat schedule that's namespaced by the service it's in."""
+class CeleryBeatScheduleBuilder(CeleryBeatSchedule):
+    """Builds a celery beat schedule.
+
+    Examples:
+        ```
+        # settings.py
+        CELERY_BEAT_SCHEDULE = CeleryBeatScheduleBuilder(
+            every_5_minutes={
+                "task": "path.to.task",
+                "schedule": CeleryBeatScheduleBuilder.crontab(minute=5),
+            },
+        )
+        ```
+    """
+
+    # Shorthand for convenience.
+    crontab = crontab
+    solar = solar
 
-    for beat in beat_schedule.values():
-        beat["task"] = namespace_task(beat["task"])
+    def __init__(self, **beat_schedule: CeleryBeat):
+        for beat in beat_schedule.values():
+            beat["task"] = namespace_task(beat["task"])
 
-    return beat_schedule
+        super().__init__(beat_schedule)
 
 
 def namespace_task(task: t.Union[str, t.Callable]):