Skip to content

Commit 0fd058c

Browse files
committed
relocate class CeleryBeat
1 parent a05db2b commit 0fd058c

File tree

3 files changed

+28
-17
lines changed

3 files changed

+28
-17
lines changed

codeforlife/settings/third_party.py

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -7,7 +7,7 @@
77

88
import typing as t
99

10-
from ..types import CeleryBeat
10+
from ..tasks import CeleryBeat
1111
from .custom import REDIS_URL, SERVICE_SITE_URL
1212
from .django import ENV
1313

codeforlife/tasks.py

Lines changed: 27 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -10,6 +10,33 @@
1010
from celery import shared_task as _shared_task
1111
from django.conf import settings
1212

13+
from .types import Args, KwArgs
14+
15+
if t.TYPE_CHECKING:
16+
from celery.schedules import crontab, solar
17+
18+
19+
class CeleryBeat(t.Dict[str, t.Any]):
20+
"""A Celery beat schedule.
21+
22+
https://docs.celeryq.dev/en/v5.4.0/userguide/periodic-tasks.html
23+
"""
24+
25+
def __init__(
26+
self,
27+
task: str,
28+
schedule: t.Union[int, "crontab", "solar"],
29+
args: t.Optional[Args] = None,
30+
kwargs: t.Optional[KwArgs] = None,
31+
):
32+
super().__init__()
33+
self["task"] = task
34+
self["schedule"] = schedule
35+
if args:
36+
self["args"] = args
37+
if kwargs:
38+
self["kwargs"] = kwargs
39+
1340

1441
def shared_task(*args, **kwargs):
1542
"""

codeforlife/types.py

Lines changed: 0 additions & 16 deletions
Original file line numberDiff line numberDiff line change
@@ -7,10 +7,6 @@
77

88
import typing as t
99

10-
if t.TYPE_CHECKING:
11-
from celery.schedules import crontab, solar
12-
13-
1410
CookieSamesite = t.Optional[t.Literal["Lax", "Strict", "None", False]]
1511

1612
Env = t.Literal["local", "development", "staging", "production"]
@@ -37,18 +33,6 @@
3733
]
3834

3935

40-
class CeleryBeat(t.NamedTuple):
41-
"""A Celery beat schedule.
42-
43-
https://docs.celeryq.dev/en/v5.4.0/userguide/periodic-tasks.html
44-
"""
45-
46-
task: str
47-
schedule: t.Union[int, "crontab", "solar"]
48-
args: t.Optional[Args] = tuple()
49-
kwargs: t.Optional[KwArgs] = {}
50-
51-
5236
def get_arg(cls: t.Type[t.Any], index: int, orig_base: int = 0):
5337
"""Get a type arg from a class.
5438

0 commit comments

Comments
 (0)