0
# Schedule Models
1
2
Database models that define when periodic tasks should execute, supporting multiple scheduling patterns including intervals, cron expressions, solar events, and one-time execution.
3
4
## Capabilities
5
6
### Interval Scheduling
7
8
Schedule tasks to run at regular intervals with configurable period types.
9
10
```python { .api }
11
class IntervalSchedule(models.Model):
12
"""
13
Schedule that runs at a specific interval.
14
15
Fields:
16
- every: int - Number of periods between runs (minimum 1)
17
- period: str - Type of period (DAYS, HOURS, MINUTES, SECONDS, MICROSECONDS)
18
"""
19
every: int
20
period: str
21
22
# Period constants
23
DAYS = 'days'
24
HOURS = 'hours'
25
MINUTES = 'minutes'
26
SECONDS = 'seconds'
27
MICROSECONDS = 'microseconds'
28
29
PERIOD_CHOICES = [
30
(DAYS, 'Days'),
31
(HOURS, 'Hours'),
32
(MINUTES, 'Minutes'),
33
(SECONDS, 'Seconds'),
34
(MICROSECONDS, 'Microseconds'),
35
]
36
37
@property
38
def schedule(self) -> schedules.schedule: ...
39
40
@property
41
def period_singular(self) -> str: ...
42
43
@classmethod
44
def from_schedule(cls, schedule: schedules.schedule) -> 'IntervalSchedule': ...
45
```
46
47
**Usage Example:**
48
49
```python
50
from django_celery_beat.models import IntervalSchedule
51
52
# Create a schedule that runs every 30 minutes
53
schedule = IntervalSchedule.objects.create(
54
every=30,
55
period=IntervalSchedule.MINUTES
56
)
57
58
# Or get existing schedule
59
schedule, created = IntervalSchedule.objects.get_or_create(
60
every=5,
61
period=IntervalSchedule.SECONDS
62
)
63
```
64
65
### Cron-style Scheduling
66
67
Timezone-aware cron-like scheduling with support for complex timing patterns.
68
69
```python { .api }
70
class CrontabSchedule(models.Model):
71
"""
72
Cron-like schedule with timezone support.
73
74
Fields follow cron format: minute hour day_of_month month_of_year day_of_week
75
Each field accepts cron expressions (*, numbers, ranges, lists).
76
"""
77
minute: str # 0-59, max 240 chars, with validators.minute_validator
78
hour: str # 0-23, max 96 chars, with validators.hour_validator
79
day_of_week: str # 0-6 (Sunday=0), max 64 chars, with validators.day_of_week_validator
80
day_of_month: str # 1-31, max 124 chars, with validators.day_of_month_validator
81
month_of_year: str # 1-12, max 64 chars, with validators.month_of_year_validator
82
timezone: timezone_field.TimeZoneField
83
84
@property
85
def schedule(self) -> Union[TzAwareCrontab, crontab]: ...
86
87
@property
88
def human_readable(self) -> str: ...
89
90
@classmethod
91
def from_schedule(cls, schedule: crontab) -> 'CrontabSchedule': ...
92
93
def due_start_time(self, initial_start_time: datetime.datetime, tz: Optional[tzinfo.tzinfo] = None) -> datetime.datetime: ...
94
```
95
96
**Usage Example:**
97
98
```python
99
from django_celery_beat.models import CrontabSchedule
100
import zoneinfo
101
102
# Run at 2:30 AM every day in Pacific timezone
103
schedule = CrontabSchedule.objects.create(
104
minute='30',
105
hour='2',
106
day_of_week='*',
107
day_of_month='*',
108
month_of_year='*',
109
timezone=zoneinfo.ZoneInfo('America/Los_Angeles')
110
)
111
112
# Run every 15 minutes
113
schedule = CrontabSchedule.objects.create(
114
minute='*/15',
115
hour='*',
116
day_of_week='*',
117
day_of_month='*',
118
month_of_year='*',
119
timezone=zoneinfo.ZoneInfo('UTC')
120
)
121
```
122
123
### Solar Event Scheduling
124
125
Schedule tasks based on astronomical events like sunrise, sunset, and solar noon for specific geographic locations.
126
127
```python { .api }
128
class SolarSchedule(models.Model):
129
"""
130
Schedule following astronomical patterns.
131
132
Fields:
133
- event: str - Solar event type (sunrise, sunset, solar_noon, etc.)
134
- latitude: decimal.Decimal - Geographic latitude (-90 to 90)
135
- longitude: decimal.Decimal - Geographic longitude (-180 to 180)
136
"""
137
event: str
138
latitude: decimal.Decimal
139
longitude: decimal.Decimal
140
141
# Solar event choices
142
SOLAR_SCHEDULES = [
143
('dawn_astronomical', 'Astronomical dawn'),
144
('dawn_civil', 'Civil dawn'),
145
('dawn_nautical', 'Nautical dawn'),
146
('dusk_astronomical', 'Astronomical dusk'),
147
('dusk_civil', 'Civil dusk'),
148
('dusk_nautical', 'Nautical dusk'),
149
('solar_noon', 'Solar noon'),
150
('sunrise', 'Sunrise'),
151
('sunset', 'Sunset'),
152
]
153
154
@property
155
def schedule(self) -> schedules.solar: ...
156
157
@classmethod
158
def from_schedule(cls, schedule: schedules.solar) -> 'SolarSchedule': ...
159
```
160
161
**Usage Example:**
162
163
```python
164
from django_celery_beat.models import SolarSchedule
165
from decimal import Decimal
166
167
# Run at sunrise in New York City
168
schedule = SolarSchedule.objects.create(
169
event='sunrise',
170
latitude=Decimal('40.7128'),
171
longitude=Decimal('-74.0060') # Note: negative for west longitude
172
)
173
174
# Run at sunset in London
175
schedule = SolarSchedule.objects.create(
176
event='sunset',
177
latitude=Decimal('51.5074'),
178
longitude=Decimal('-0.1278')
179
)
180
```
181
182
### One-time Scheduling
183
184
Schedule tasks to run exactly once at a specific date and time.
185
186
```python { .api }
187
class ClockedSchedule(models.Model):
188
"""
189
Schedule for one-time execution at a specific datetime.
190
191
Fields:
192
- clocked_time: datetime.datetime - When the task should run
193
"""
194
clocked_time: datetime.datetime
195
196
@property
197
def schedule(self) -> clocked: ...
198
199
@classmethod
200
def from_schedule(cls, schedule: clocked) -> 'ClockedSchedule': ...
201
```
202
203
**Usage Example:**
204
205
```python
206
from django_celery_beat.models import ClockedSchedule
207
from datetime import datetime, timedelta
208
209
# Run once in 1 hour
210
future_time = datetime.utcnow() + timedelta(hours=1)
211
schedule = ClockedSchedule.objects.create(
212
clocked_time=future_time
213
)
214
215
# Run at a specific time
216
specific_time = datetime(2024, 12, 25, 9, 0, 0) # Christmas 9 AM
217
schedule = ClockedSchedule.objects.create(
218
clocked_time=specific_time
219
)
220
```
221
222
## Model Helper Functions
223
224
### Cron Expression Formatting
225
226
```python { .api }
227
def cronexp(field: Optional[str]) -> str:
228
"""
229
Convert cron field to string representation.
230
231
Parameters:
232
- field: Cron field value or None
233
234
Returns:
235
str: String representation, '*' if field is None/empty
236
"""
237
```
238
239
### Timezone Detection
240
241
```python { .api }
242
def crontab_schedule_celery_timezone() -> str:
243
"""
244
Get timezone string from Django CELERY_TIMEZONE setting.
245
246
Returns:
247
str: Timezone string, defaults to 'UTC' if not set or invalid
248
"""
249
```
250
251
## Constants
252
253
### Period Constants
254
255
```python { .api }
256
DAYS = 'days'
257
HOURS = 'hours'
258
MINUTES = 'minutes'
259
SECONDS = 'seconds'
260
MICROSECONDS = 'microseconds'
261
262
PERIOD_CHOICES = [
263
(DAYS, 'Days'),
264
(HOURS, 'Hours'),
265
(MINUTES, 'Minutes'),
266
(SECONDS, 'Seconds'),
267
(MICROSECONDS, 'Microseconds'),
268
]
269
270
SINGULAR_PERIODS = [
271
(DAYS, 'Day'),
272
(HOURS, 'Hour'),
273
(MINUTES, 'Minute'),
274
(SECONDS, 'Second'),
275
(MICROSECONDS, 'Microsecond'),
276
]
277
```
278
279
### Solar Event Constants
280
281
```python { .api }
282
SOLAR_SCHEDULES = [
283
('dawn_astronomical', 'Astronomical dawn'),
284
('dawn_civil', 'Civil dawn'),
285
('dawn_nautical', 'Nautical dawn'),
286
('dusk_astronomical', 'Astronomical dusk'),
287
('dusk_civil', 'Civil dusk'),
288
('dusk_nautical', 'Nautical dusk'),
289
('solar_noon', 'Solar noon'),
290
('sunrise', 'Sunrise'),
291
('sunset', 'Sunset'),
292
]
293
```
294
295
## Types
296
297
### Custom Schedule Types
298
299
```python { .api }
300
class clocked(BaseSchedule):
301
"""
302
Schedule for one-time execution at specific datetime.
303
"""
304
def __init__(self, clocked_time: datetime.datetime): ...
305
def is_due(self, last_run_at: datetime.datetime) -> tuple[bool, float]: ...
306
def remaining_estimate(self, last_run_at: datetime.datetime) -> datetime.timedelta: ...
307
308
class TzAwareCrontab(crontab):
309
"""
310
Timezone-aware cron scheduling.
311
"""
312
def __init__(self, minute='*', hour='*', day_of_week='*',
313
day_of_month='*', month_of_year='*',
314
tz: Optional[tzinfo.tzinfo] = None): ...
315
def is_due(self, last_run_at: datetime.datetime) -> tuple[bool, float]: ...
316
def nowfunc(self) -> datetime.datetime: ...
317
```