9-09 15 views
一、背景
用过django的人应该都有被其强大、特方便、特丰富的管理后台所震撼，在安装好celery、django-celery模块后，会内置一套task的管理页面，如下图：
django-admin的celery页面
crontab配置页面
periodic tasks 配置页面
二、实现
但是这都是django-admin模板中的，我的需求是将这个功能跟自己的运维平台整合到一起，面对这个需求主要是操作两个模型，crontab和periodictask
1. Crontab
Crontab比较简单，就一张表，其实直接写个ModelForm去操作他的模型就可以了
lib/python3.6/site-packages/django_celery_beat/models.py
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 |
@python_2_unicode_compatible class CrontabSchedule(models.Model): """Crontab-like schedule.""" # # The worst case scenario for day of month is a list of all 31 day numbers # '[1, 2, ..., 31]' which has a length of 115. Likewise, minute can be # 0..59 and hour can be 0..23. Ensure we can accomodate these by allowing # 4 chars for each value (what we save on 0-9 accomodates the []). # We leave the other fields at their historical length. # minute = models.CharField(_('minute'), max_length=60 * 4, default='*') hour = models.CharField(_('hour'), max_length=24 * 4, default='*') day_of_week = models.CharField( _('day of week'), max_length=64, default='*', ) day_of_month = models.CharField( _('day of month'), max_length=31 * 4, default='*', ) month_of_year = models.CharField( _('month of year'), max_length=64, default='*', ) class Meta: """Table information.""" verbose_name = _('crontab') verbose_name_plural = _('crontabs') ordering = ['month_of_year', 'day_of_month', 'day_of_week', 'hour', 'minute'] def __str__(self): return '{0} {1} {2} {3} {4} (m/h/d/dM/MY)'.format( cronexp(self.minute), cronexp(self.hour), cronexp(self.day_of_week), cronexp(self.day_of_month), cronexp(self.month_of_year), ) @property def schedule(self): return schedules.crontab(minute=self.minute, hour=self.hour, day_of_week=self.day_of_week, day_of_month=self.day_of_month, month_of_year=self.month_of_year, nowfun=lambda: make_aware(now())) @classmethod def from_schedule(cls, schedule): spec = {'minute': schedule._orig_minute, 'hour': schedule._orig_hour, 'day_of_week': schedule._orig_day_of_week, 'day_of_month': schedule._orig_day_of_month, 'month_of_year': schedule._orig_month_of_year} try: return cls.objects.get(**spec) except cls.DoesNotExist: return cls(**spec) except MultipleObjectsReturned: cls.objects.filter(**spec).delete() return cls(**spec) |
2. Periodic Task
Periodic Task稍微复杂一点，因为涉及到register task，不过经分析django-admin的中Periodic Task其时是通过“lib/python3.6/site-packages/django_celery_beat/admin.py”中的“PeriodicTaskForm”去操作“PeriodicTask”这个模型的，其实我们可以直接调用这个已经写的好的ModelForm
lib/python3.6/site-packages/django_celery_beat/admin.py
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 |
class PeriodicTaskForm(forms.ModelForm): """Form that lets you create and modify periodic tasks.""" regtask = TaskChoiceField( label=_('Task (registered)'), required=False, ) task = forms.CharField( label=_('Task (custom)'), required=False, max_length=200, ) class Meta: """Form metadata.""" model = PeriodicTask exclude = () def clean(self): data = super(PeriodicTaskForm, self).clean() regtask = data.get('regtask') if regtask: data['task'] = regtask if not data['task']: exc = forms.ValidationError(_('Need name of task')) self._errors['task'] = self.error_class(exc.messages) raise exc return data def _clean_json(self, field): value = self.cleaned_data[field] try: loads(value) except ValueError as exc: raise forms.ValidationError( _('Unable to parse JSON: %s') % exc, ) return value def clean_args(self): return self._clean_json('args') def clean_kwargs(self): return self._clean_json('kwargs') |
2.1 Crontab
2.1.1 Crontab ModelForm
有了上面的线索，我们现在来写Crontab的Form
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 |
#!/usr/bin/env python3 # -*- coding: utf-8 -*- # @Author : Eric Winn # @Email : eng.eric.winn@gmail.com # @Time : 2018-05-17 17:06 # @Version : 2.0 # @File : forms # @Software : PyCharm from __future__ import absolute_import, unicode_literals from django import forms from django_celery_beat.models import CrontabSchedule class CrontabScheduleForm(forms.ModelForm): class Meta: model = CrontabSchedule fields = '__all__' |
2.1.2 Crontab Create&Update View
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 |
from django_celery_beat.models import CrontabSchedule from .forms import CrontabScheduleForm # Crontab Create View class CrontabScheduleCreateView(AdminUserRequiredMixin, SuccessMessageMixin, CreateView): model = CrontabSchedule form_class = CrontabScheduleForm template_name = 'ops/crontab_create.html' success_url = reverse_lazy('ops:crontab-list') def get_form(self, form_class=None): form = super().get_form(form_class=form_class) return form def get_context_data(self, **kwargs): context = { 'app': _('Crontab'), 'action': _('Create Crontab'), } kwargs.update(context) return super().get_context_data(**kwargs) def get_success_message(self, cleaned_data): return create_success_msg % ({"name": '{} {} {} {} {}'.format(*[ cleaned_data[field.name] for field in CrontabSchedule._meta.fields if field.name not in [ 'id', ] ])}) # Crontab Update View class CrontabScheduleUpdateView(AdminUserRequiredMixin, SuccessMessageMixin, UpdateView): model = CrontabSchedule form_class = CrontabScheduleForm template_name = 'ops/crontab_update.html' success_url = reverse_lazy('ops:crontab-list') def get_context_data(self, **kwargs): context = { 'app': _('Crontab'), 'action': _('Update Crontab'), } kwargs.update(context) return super().get_context_data(**kwargs) def get_success_message(self, cleaned_data): return create_success_msg % ({"name": '{} {} {} {} {}'.format(*[ cleaned_data[field.name] for field in CrontabSchedule._meta.fields if field.name not in [ 'id', ] ])}) |
2.1.3 Crontab Url
|
1 2 3 4 |
urlpatterns = [ url(r'^crontab/create/$', views.CrontabScheduleCreateView.as_view(), name='crontab-create'), url(r'^crontab/(?P<pk>\d+)/update/$', views.CrontabUpdateView.as_view(), name='crontab-update'), ] |
2.1.4 Crontab Create&Update Template
Create和Update共用同一个template
|
1 2 3 4 5 |
{% bootstrap_field form.minute layout="horizontal" %} {% bootstrap_field form.hour layout="horizontal" %} {% bootstrap_field form.day_of_week layout="horizontal" %} {% bootstrap_field form.day_of_month layout="horizontal" %} {% bootstrap_field form.month_of_year layout="horizontal" %} |
2.1.5 页面效果
2.2 Periodic Task
有了上面的线索，我们直接调用现成的就行
2.2.1 Periodic Task Create&Update View
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 |
from django_celery_beat.admin import PeriodicTaskForm from django_celery_beat.models import PeriodicTask # PeriodicTask Create View class TaskCreateView(AdminUserRequiredMixin, SuccessMessageMixin, CreateView): model = PeriodicTask form_class = PeriodicTaskForm template_name = 'ops/task_create.html' success_url = reverse_lazy('ops:task-list') def get_form(self, form_class=None): form = super().get_form(form_class=form_class) return form def get_context_data(self, **kwargs): context = { 'app': _('Task'), 'action': _('Create Task'), } kwargs.update(context) return super().get_context_data(**kwargs) def get_success_message(self, cleaned_data): return create_success_msg % ({"name": cleaned_data["name"]}) # PeriodicTask Update View class TaskUpdateView(AdminUserRequiredMixin, SuccessMessageMixin, UpdateView): model = PeriodicTask form_class = PeriodicTaskForm template_name = 'ops/task_update.html' success_url = reverse_lazy('ops:task-list') def get_context_data(self, **kwargs): context = { 'app': _('Task'), 'action': _('Update Task'), } kwargs.update(context) return super().get_context_data(**kwargs) def get_success_message(self, cleaned_data): return update_success_msg % ({"name": cleaned_data["name"]}) |
2.2.2 Periodic Task Url
|
1 2 3 4 |
urlpatterns = [ url(r'^task/create/$', views.TaskCreateView.as_view(), name='task-create'), url(r'^task/(?P<pk>\d+)/update/$', views.TaskUpdateView.as_view(), name='task-update'), ] |
2.2.3 Periodic Task Create&Update Template
Create和Update共用同一个template
|
1 2 3 4 5 6 7 |
{% bootstrap_field form.name layout="horizontal" %} {% bootstrap_field form.regtask layout="horizontal" %} {% bootstrap_field form.task layout="horizontal" %} {% bootstrap_field form.enabled layout="horizontal" %} {% bootstrap_field form.crontab layout="horizontal" %} {% bootstrap_field form.args layout="horizontal" %} {% bootstrap_field form.kwargs layout="horizontal" %} |
2.2.4 页面效果
如果想赏钱，可以用微信扫描下面的二维码，一来能刺激我写博客的欲望，二来好维护云主机的费用； 另外再次标注博客原地址 itnotebooks.com 感谢！
请问periodictask的注册函数列表是怎么获取的？文章并没有提到，请问有源码吗？
celery包中有个current_app的tasks属性是获取当前项目中所有注册的task，另外建议对页面没有太多自定义的内容的话，完全可以直接使用或继成celery包中Form类“from django_celery_beat.admin import PeriodicTaskForm“