0
# Admin Interface
1
2
Django Admin integration providing web-based management interface for periodic tasks and schedules with custom forms, actions, validation, and task execution capabilities.
3
4
## Capabilities
5
6
### Periodic Task Admin
7
8
Main admin interface for managing periodic tasks with custom actions and forms.
9
10
```python { .api }
11
class PeriodicTaskAdmin(ModelAdmin):
12
"""
13
Django Admin interface for PeriodicTask model.
14
15
Provides form-based task management with custom widgets,
16
bulk actions, and task execution capabilities.
17
"""
18
form = PeriodicTaskForm
19
model = PeriodicTask
20
21
# Admin actions
22
def enable_tasks(self, request, queryset): ...
23
def disable_tasks(self, request, queryset): ...
24
def toggle_tasks(self, request, queryset): ...
25
def run_tasks(self, request, queryset): ...
26
27
# Custom admin methods
28
def get_queryset(self, request): ...
29
def save_model(self, request, obj, form, change): ...
30
```
31
32
**Available Admin Actions:**
33
34
- **Enable Tasks**: Enable selected periodic tasks
35
- **Disable Tasks**: Disable selected periodic tasks
36
- **Toggle Tasks**: Toggle enabled/disabled state of selected tasks
37
- **Run Tasks**: Manually execute selected tasks immediately
38
39
**Usage Examples:**
40
41
```python
42
# Admin actions are available in the Django Admin interface
43
# Select tasks and use the action dropdown to:
44
45
# 1. Enable multiple tasks at once
46
# - Select disabled tasks
47
# - Choose "Enable selected periodic tasks" from actions dropdown
48
49
# 2. Disable tasks for maintenance
50
# - Select enabled tasks
51
# - Choose "Disable selected periodic tasks" from actions dropdown
52
53
# 3. Toggle task states
54
# - Select any tasks
55
# - Choose "Toggle selected periodic tasks" from actions dropdown
56
57
# 4. Run tasks immediately (for testing)
58
# - Select tasks to execute
59
# - Choose "Run selected periodic tasks" from actions dropdown
60
# - Tasks will be sent to Celery workers immediately
61
```
62
63
### Schedule Admin Classes
64
65
Admin interfaces for managing different schedule types.
66
67
```python { .api }
68
class CrontabScheduleAdmin(ModelAdmin):
69
"""
70
Admin interface for CrontabSchedule with human-readable display.
71
"""
72
list_display = ('__str__', 'timezone', 'human_readable')
73
list_filter = ('timezone',)
74
search_fields = ('minute', 'hour', 'day_of_week', 'day_of_month', 'month_of_year')
75
76
class IntervalScheduleAdmin(ModelAdmin):
77
"""
78
Admin interface for IntervalSchedule.
79
"""
80
list_display = ('__str__', 'every', 'period')
81
list_filter = ('period',)
82
search_fields = ('every',)
83
84
class SolarScheduleAdmin(ModelAdmin):
85
"""
86
Admin interface for SolarSchedule.
87
"""
88
list_display = ('__str__', 'event', 'latitude', 'longitude')
89
list_filter = ('event',)
90
search_fields = ('latitude', 'longitude')
91
92
class ClockedScheduleAdmin(ModelAdmin):
93
"""
94
Admin interface for ClockedSchedule.
95
"""
96
list_display = ('__str__', 'clocked_time')
97
list_filter = ('clocked_time',)
98
99
class PeriodicTaskInline(TabularInline):
100
"""
101
Inline admin interface for displaying PeriodicTasks within schedule admins.
102
"""
103
model = PeriodicTask
104
fields = ('name', 'task', 'args', 'kwargs')
105
readonly_fields = ('name', 'task', 'args', 'kwargs')
106
can_delete: bool
107
extra: int
108
show_change_link: bool
109
110
def has_add_permission(self, request, obj: Optional[Model] = None) -> bool: ...
111
```
112
113
### Custom Forms and Widgets
114
115
Specialized form components for task and schedule management.
116
117
```python { .api }
118
class TaskSelectWidget(Select):
119
"""
120
Widget for selecting Celery tasks from registered task list.
121
122
Dynamically populates choices with available Celery tasks.
123
"""
124
celery_app: Celery # Current Celery app instance
125
_choices: Optional[tuple] # Cached task choices
126
127
def tasks_as_choices(self) -> tuple[tuple[str, str], ...]: ...
128
129
@property
130
def choices(self) -> tuple: ...
131
132
@choices.setter
133
def choices(self, _) -> None: ...
134
135
@cached_property
136
def _modules(self): ...
137
138
class TaskChoiceField(ChoiceField):
139
"""
140
Form field for selecting Celery tasks with validation.
141
142
Validates that selected task is registered with Celery.
143
"""
144
widget = TaskSelectWidget
145
146
def __init__(self, *args, **kwargs): ...
147
148
def valid_value(self, value: str) -> bool: ...
149
150
class PeriodicTaskForm(ModelForm):
151
"""
152
ModelForm for PeriodicTask with custom validation and widgets.
153
154
Provides enhanced task selection, JSON validation, and schedule validation.
155
"""
156
task = TaskChoiceField()
157
158
class Meta:
159
model = PeriodicTask
160
fields = '__all__'
161
162
def clean(self) -> dict: ...
163
def clean_args(self) -> str: ...
164
def clean_kwargs(self) -> str: ...
165
def clean_headers(self) -> str: ...
166
```
167
168
**Usage Examples:**
169
170
```python
171
# Custom form usage in Django Admin (automatic)
172
# The PeriodicTaskForm provides:
173
174
# 1. Task selection dropdown
175
# - Automatically populated with registered Celery tasks
176
# - Validates task exists in Celery registry
177
178
# 2. JSON validation for arguments
179
# - Validates args field contains valid JSON list
180
# - Validates kwargs field contains valid JSON object
181
# - Validates headers field contains valid JSON object
182
183
# 3. Schedule validation
184
# - Ensures exactly one schedule type is selected
185
# - Validates schedule parameters are correct
186
187
# Example form data validation:
188
form_data = {
189
'name': 'Test Task',
190
'task': 'myapp.tasks.example_task', # Must be registered with Celery
191
'args': '["arg1", "arg2"]', # Must be valid JSON list
192
'kwargs': '{"param": "value"}', # Must be valid JSON object
193
'headers': '{"priority": 9}', # Must be valid JSON object
194
'interval': interval_schedule.id, # Must select exactly one schedule
195
'enabled': True
196
}
197
198
form = PeriodicTaskForm(data=form_data)
199
if form.is_valid():
200
task = form.save()
201
else:
202
print("Validation errors:", form.errors)
203
```
204
205
## Admin Interface Features
206
207
### Task Management Interface
208
209
The Django Admin provides a comprehensive interface for task management:
210
211
**List View Features:**
212
- Task name, task function, schedule type, and status
213
- Filtering by enabled status, schedule type, and queue
214
- Search by task name and description
215
- Bulk actions for enable/disable/toggle/run operations
216
217
**Detail View Features:**
218
- Organized fieldsets for task identification, scheduling, arguments, routing, and tracking
219
- Custom task selection widget with autocomplete
220
- JSON syntax highlighting and validation for arguments
221
- Schedule association with validation
222
- Execution history display (last run time, total runs)
223
224
### Schedule Management Interface
225
226
Each schedule type has its own admin interface:
227
228
**Crontab Schedule Admin:**
229
- Human-readable cron description display
230
- Timezone filtering and search
231
- Cron field validation with helpful error messages
232
233
**Interval Schedule Admin:**
234
- Period type filtering (days, hours, minutes, seconds, microseconds)
235
- Simple every/period display
236
237
**Solar Schedule Admin:**
238
- Solar event filtering (sunrise, sunset, etc.)
239
- Geographic coordinate search and validation
240
241
**Clocked Schedule Admin:**
242
- Date/time filtering and search
243
- One-time execution tracking
244
245
### Admin Customization
246
247
```python
248
# Custom admin registration with extended functionality
249
from django.contrib import admin
250
from django_celery_beat.models import PeriodicTask, CrontabSchedule
251
from django_celery_beat.admin import PeriodicTaskAdmin, CrontabScheduleAdmin
252
253
class CustomPeriodicTaskAdmin(PeriodicTaskAdmin):
254
"""Extended admin with custom features."""
255
256
list_display = PeriodicTaskAdmin.list_display + ('total_run_count', 'last_run_at')
257
list_filter = PeriodicTaskAdmin.list_filter + ('date_changed',)
258
readonly_fields = ('total_run_count', 'last_run_at', 'date_changed')
259
260
def get_queryset(self, request):
261
"""Add custom filtering or optimization."""
262
qs = super().get_queryset(request)
263
# Add custom filters or select_related optimizations
264
return qs.select_related('interval', 'crontab', 'solar', 'clocked')
265
266
def save_model(self, request, obj, form, change):
267
"""Add custom save logic."""
268
if not change: # New task
269
obj.description = f"Created by {request.user.username}"
270
super().save_model(request, obj, form, change)
271
272
# Re-register with custom admin
273
admin.site.unregister(PeriodicTask)
274
admin.site.register(PeriodicTask, CustomPeriodicTaskAdmin)
275
```
276
277
### Admin Templates
278
279
The package includes custom admin templates for enhanced functionality:
280
281
**Custom Change Form Template:**
282
- JavaScript for cron expression human-readable translation
283
- Enhanced task selection interface
284
- Real-time JSON validation feedback
285
286
**Custom Change List Template:**
287
- Enhanced bulk action feedback
288
- Task execution status indicators
289
- Quick enable/disable toggles
290
291
## Programmatic Admin Usage
292
293
### Admin Actions via Code
294
295
```python
296
from django_celery_beat.admin import PeriodicTaskAdmin
297
from django_celery_beat.models import PeriodicTask
298
from django.contrib.admin.sites import site
299
from django.http import HttpRequest
300
301
# Simulate admin actions programmatically
302
admin_instance = PeriodicTaskAdmin(PeriodicTask, site)
303
request = HttpRequest() # Mock request object
304
305
# Enable tasks
306
queryset = PeriodicTask.objects.filter(enabled=False)
307
admin_instance.enable_tasks(request, queryset)
308
309
# Disable tasks
310
queryset = PeriodicTask.objects.filter(name__startswith='test_')
311
admin_instance.disable_tasks(request, queryset)
312
313
# Run tasks immediately
314
queryset = PeriodicTask.objects.filter(name='urgent_task')
315
admin_instance.run_tasks(request, queryset)
316
```
317
318
### Custom Admin Views
319
320
```python
321
from django.contrib import admin
322
from django.urls import path
323
from django.shortcuts import render
324
from django.http import JsonResponse
325
from django_celery_beat.models import PeriodicTask
326
327
class ExtendedPeriodicTaskAdmin(PeriodicTaskAdmin):
328
"""Admin with custom views for advanced task management."""
329
330
def get_urls(self):
331
"""Add custom admin URLs."""
332
urls = super().get_urls()
333
custom_urls = [
334
path('task-status/', self.task_status_view, name='task-status'),
335
path('bulk-operations/', self.bulk_operations_view, name='bulk-operations'),
336
]
337
return custom_urls + urls
338
339
def task_status_view(self, request):
340
"""Custom view for task status overview."""
341
stats = {
342
'total_tasks': PeriodicTask.objects.count(),
343
'enabled_tasks': PeriodicTask.objects.filter(enabled=True).count(),
344
'disabled_tasks': PeriodicTask.objects.filter(enabled=False).count(),
345
'one_off_tasks': PeriodicTask.objects.filter(one_off=True).count(),
346
}
347
348
if request.headers.get('Accept') == 'application/json':
349
return JsonResponse(stats)
350
351
return render(request, 'admin/task_status.html', {'stats': stats})
352
353
def bulk_operations_view(self, request):
354
"""Custom view for bulk task operations."""
355
if request.method == 'POST':
356
operation = request.POST.get('operation')
357
task_ids = request.POST.getlist('task_ids')
358
359
queryset = PeriodicTask.objects.filter(id__in=task_ids)
360
361
if operation == 'enable':
362
queryset.update(enabled=True)
363
elif operation == 'disable':
364
queryset.update(enabled=False)
365
elif operation == 'delete':
366
queryset.delete()
367
368
return JsonResponse({'status': 'success', 'affected': len(task_ids)})
369
370
tasks = PeriodicTask.objects.all()
371
return render(request, 'admin/bulk_operations.html', {'tasks': tasks})
372
```
373
374
## Error Handling and Validation
375
376
### Form Validation Errors
377
378
```python
379
from django_celery_beat.admin import PeriodicTaskForm
380
from django.core.exceptions import ValidationError
381
382
# Common validation scenarios
383
def handle_form_validation():
384
# Invalid JSON in args
385
form_data = {
386
'name': 'Test Task',
387
'task': 'myapp.tasks.test',
388
'args': 'invalid json', # Will cause validation error
389
'interval': schedule.id
390
}
391
392
form = PeriodicTaskForm(data=form_data)
393
if not form.is_valid():
394
print("Args validation error:", form.errors['args'])
395
396
# Missing schedule
397
form_data = {
398
'name': 'Test Task',
399
'task': 'myapp.tasks.test',
400
# No schedule specified - will cause validation error
401
}
402
403
form = PeriodicTaskForm(data=form_data)
404
if not form.is_valid():
405
print("Schedule validation error:", form.non_field_errors())
406
407
# Invalid task name
408
form_data = {
409
'name': 'Test Task',
410
'task': 'nonexistent.task', # Task not registered with Celery
411
'interval': schedule.id
412
}
413
414
form = PeriodicTaskForm(data=form_data)
415
if not form.is_valid():
416
print("Task validation error:", form.errors['task'])
417
```
418
419
### Admin Action Error Handling
420
421
```python
422
from django.contrib import messages
423
from django.contrib.admin import ModelAdmin
424
425
class RobustPeriodicTaskAdmin(PeriodicTaskAdmin):
426
"""Admin with enhanced error handling."""
427
428
def run_tasks(self, request, queryset):
429
"""Run tasks with error handling."""
430
success_count = 0
431
error_count = 0
432
433
for task in queryset:
434
try:
435
# Send task to Celery
436
from celery import current_app
437
current_app.send_task(
438
task.task,
439
args=task.args,
440
kwargs=task.kwargs,
441
queue=task.queue,
442
routing_key=task.routing_key,
443
priority=task.priority
444
)
445
success_count += 1
446
except Exception as e:
447
error_count += 1
448
messages.error(request, f"Failed to run task {task.name}: {e}")
449
450
if success_count:
451
messages.success(request, f"Successfully queued {success_count} tasks")
452
if error_count:
453
messages.error(request, f"Failed to queue {error_count} tasks")
454
455
run_tasks.short_description = "Run selected periodic tasks (with error handling)"
456
```