0
# Task Scheduling and Periodic Tasks
1
2
Advanced scheduling capabilities including delayed execution, periodic tasks with cron-like syntax, and task pipeline management. These features enable sophisticated task orchestration and time-based automation.
3
4
## Capabilities
5
6
### Periodic Task Creation
7
8
Create tasks that run automatically on a schedule defined by cron-like expressions.
9
10
```python { .api }
11
def periodic_task(self, validate_datetime, retries=0, retry_delay=0,
12
priority=None, context=False, name=None, expires=None, **kwargs):
13
"""
14
Decorator to create a periodic task.
15
16
Parameters:
17
- validate_datetime: Function that returns True when task should run
18
- retries (int): Number of retry attempts (default: 0)
19
- retry_delay (int): Delay between retries in seconds (default: 0)
20
- priority (int): Task priority (optional)
21
- context (bool): Pass task instance as keyword argument (default: False)
22
- name (str): Custom task name (default: function name)
23
- expires (datetime/int): Task expiration (optional)
24
- **kwargs: Additional task configuration
25
26
Returns:
27
TaskWrapper for periodic task
28
"""
29
```
30
31
### Cron-style Scheduling
32
33
Create cron-like schedule validation functions for precise timing control.
34
35
```python { .api }
36
def crontab(minute='*', hour='*', day='*', month='*', day_of_week='*', strict=False):
37
"""
38
Create a crontab-style schedule validator.
39
40
Parameters:
41
- minute (str): Minute specification (0-59, default: '*')
42
- hour (str): Hour specification (0-23, default: '*')
43
- day (str): Day of month specification (1-31, default: '*')
44
- month (str): Month specification (1-12, default: '*')
45
- day_of_week (str): Day of week specification (0-6, 0=Sunday, default: '*')
46
- strict (bool): Raise ValueError for invalid formats (default: False)
47
48
Supported formats:
49
- '*': Every value
50
- 'n': Specific value
51
- 'm,n': Multiple values
52
- 'm-n': Range of values
53
- '*/n': Every nth value
54
55
Returns:
56
Function that validates datetime against cron pattern
57
"""
58
```
59
60
### Task Scheduling
61
62
Schedule individual tasks for delayed execution.
63
64
```python { .api }
65
def schedule(self, args=None, kwargs=None, eta=None, delay=None,
66
priority=None, retries=None, retry_delay=None, expires=None, id=None):
67
"""
68
Schedule a task for future execution.
69
70
Parameters:
71
- args (tuple): Task arguments (optional)
72
- kwargs (dict): Task keyword arguments (optional)
73
- eta (datetime): Exact execution time (optional)
74
- delay (int/float/timedelta): Delay before execution (optional)
75
- priority (int): Task priority (optional)
76
- retries (int): Number of retries (optional)
77
- retry_delay (int): Delay between retries (optional)
78
- expires (datetime/int): Task expiration (optional)
79
- id (str): Custom task ID (optional)
80
81
Returns:
82
Result instance
83
"""
84
85
def add_schedule(self, task):
86
"""
87
Add a task to the schedule queue.
88
89
Parameters:
90
- task (Task): Task with eta set for future execution
91
92
Returns:
93
None
94
"""
95
```
96
97
### Schedule Management
98
99
Monitor and manage scheduled tasks.
100
101
```python { .api }
102
def read_schedule(self, timestamp=None):
103
"""
104
Read tasks ready for execution from schedule.
105
106
Parameters:
107
- timestamp (datetime): Current time (default: now)
108
109
Returns:
110
List of Task instances ready to run
111
"""
112
113
def read_periodic(self, timestamp):
114
"""
115
Get periodic tasks that should run at given time.
116
117
Parameters:
118
- timestamp (datetime): Time to check periodic tasks against
119
120
Returns:
121
List of periodic Task instances
122
"""
123
124
def scheduled(self, limit=None):
125
"""
126
Get list of scheduled tasks.
127
128
Parameters:
129
- limit (int): Maximum number of tasks to return (optional)
130
131
Returns:
132
List of scheduled Task instances
133
"""
134
135
def scheduled_count(self):
136
"""
137
Get the number of scheduled tasks.
138
139
Returns:
140
int: Number of tasks in schedule
141
"""
142
143
def ready_to_run(self, task, timestamp=None):
144
"""
145
Check if a task is ready for execution.
146
147
Parameters:
148
- task (Task): Task to check
149
- timestamp (datetime): Current time (default: now)
150
151
Returns:
152
bool: True if task should run now
153
"""
154
```
155
156
## Usage Examples
157
158
### Periodic Tasks with Cron Schedules
159
160
```python
161
from huey import RedisHuey, crontab
162
163
huey = RedisHuey('scheduler-app')
164
165
# Run every day at 2:30 AM
166
@huey.periodic_task(crontab(minute='30', hour='2'))
167
def daily_backup():
168
backup_database()
169
return "Daily backup completed"
170
171
# Run every 15 minutes
172
@huey.periodic_task(crontab(minute='*/15'))
173
def health_check():
174
check_system_health()
175
return "Health check completed"
176
177
# Run on weekdays at 9 AM
178
@huey.periodic_task(crontab(minute='0', hour='9', day_of_week='1-5'))
179
def weekday_report():
180
generate_daily_report()
181
return "Daily report generated"
182
183
# Run on the first day of each month
184
@huey.periodic_task(crontab(minute='0', hour='0', day='1'))
185
def monthly_cleanup():
186
cleanup_old_data()
187
return "Monthly cleanup completed"
188
```
189
190
### Delayed Task Execution
191
192
```python
193
import datetime
194
from huey import RedisHuey
195
196
huey = RedisHuey('delayed-tasks')
197
198
@huey.task()
199
def send_reminder(user_id, message):
200
# Send reminder logic
201
return f"Reminder sent to user {user_id}: {message}"
202
203
# Schedule for specific datetime
204
eta = datetime.datetime(2024, 1, 15, 14, 30)
205
result = send_reminder.schedule(
206
args=(123, "Your appointment is tomorrow"),
207
eta=eta
208
)
209
210
# Schedule with delay (5 minutes from now)
211
result = send_reminder.schedule(
212
args=(456, "Meeting starts in 10 minutes"),
213
delay=300 # 5 minutes
214
)
215
216
# Schedule with timedelta
217
result = send_reminder.schedule(
218
args=(789, "Weekly report due"),
219
delay=datetime.timedelta(hours=2)
220
)
221
```
222
223
### Complex Scheduling Patterns
224
225
```python
226
# Custom schedule validation function
227
def business_hours(dt):
228
"""Run only during business hours (9 AM - 5 PM, weekdays)"""
229
return (dt.hour >= 9 and dt.hour < 17 and
230
dt.weekday() < 5) # Monday=0, Sunday=6
231
232
@huey.periodic_task(business_hours)
233
def business_hours_task():
234
process_business_data()
235
return "Business hours processing complete"
236
237
# Multiple schedule patterns
238
@huey.periodic_task(crontab(minute='0', hour='*/3')) # Every 3 hours
239
def frequent_sync():
240
sync_external_data()
241
return "Data sync completed"
242
243
@huey.periodic_task(crontab(minute='0', hour='0', day_of_week='0')) # Weekly on Sunday
244
def weekly_maintenance():
245
perform_maintenance()
246
return "Weekly maintenance completed"
247
```
248
249
### Task Scheduling with Context
250
251
```python
252
@huey.task(context=True)
253
def process_file(filename, task=None):
254
# Access task metadata
255
print(f"Processing {filename} (Task ID: {task.id})")
256
257
# Simulate file processing
258
time.sleep(2)
259
return f"Processed {filename}"
260
261
# Schedule with task context
262
result = process_file.schedule(
263
args=("important_data.csv",),
264
delay=60, # Process in 1 minute
265
priority=10, # High priority
266
retries=2, # Retry twice if it fails
267
retry_delay=30 # Wait 30 seconds between retries
268
)
269
```
270
271
### Schedule Monitoring
272
273
```python
274
# Check scheduled tasks
275
scheduled_tasks = huey.scheduled(limit=10)
276
print(f"Next {len(scheduled_tasks)} scheduled tasks:")
277
for task in scheduled_tasks:
278
print(f"- {task.name} at {task.eta}")
279
280
# Check schedule size
281
print(f"Total scheduled tasks: {huey.scheduled_count()}")
282
283
# Read tasks ready to run
284
ready_tasks = huey.read_schedule()
285
print(f"Tasks ready to run: {len(ready_tasks)}")
286
```