livingbio-django-q 0.8.5

功能

• 多处理工作池
• 异步任务
• 计划任务和重复任务
• 加密和压缩包
• 失败和成功数据库或缓存
• 结果挂钩、组和链
• Django管理集成
• paas兼容多个实例
• 多群集监视器
• redis、disfe、ironmq、sqs、mongodb或orm
• 滚动条支架

要求

• Django>；=1.8
• Django-picklefield
• Arrow
• Blessed

测试工具：python 2.7&3.6。Django 1.8.18、1.10.7和1.11

经纪人

• Redis
• Disque
• IronMQ
• Amazon SQS
• MongoDB
• Django ORM

安装

• 使用pip安装最新版本：

  $ pip install django-q
  

• 将django q添加到项目设置中已安装的应用程序中。py：

  INSTALLED_APPS = (
      # other apps
      'django_q',
  )
  

• 运行django迁移以创建数据库表：

  $ python manage.py migrate
  

• 选择消息broker，配置并安装适当的客户端库。

阅读https://django-q.readthedocs.org上的完整文档

配置

所有配置设置都是可选的。例如：

# settings.py exampleQ_CLUSTER={'name':'myproject','workers':8,'recycle':500,'timeout':60,'compress':True,'cpu_affinity':1,'save_limit':250,'queue_limit':500,'label':'Django Q','redis':{'host':'127.0.0.1','port':6379,'db':0,}}

有关完整配置选项，请参见configuration documentation。

管理命令

使用以下命令启动群集：

$ python manage.py qcluster

使用以下命令监视群集：

$ python manage.py qmonitor

使用以下命令检查总体统计信息：

$ python manage.py qinfo

创建任务

使用代码中的异步来快速卸载任务：

fromdjango_q.tasksimportasync,result# create the taskasync('math.copysign',2,-2)# or with a referenceimportmath.copysigntask_id=async(copysign,2,-2)# get the resulttask_result=result(task_id)# result returns None if the task has not been executed yet# you can wait for ittask_result=result(task_id,200)# but in most cases you will want to use a hook:async('math.modf',2.5,hook='hooks.print_result')# hooks.pydefprint_result(task):print(task.result)

有关详细信息，请参见Tasks

日程

时间表是常规的django模型。你可以通过 管理页或直接从您的代码：

# Use the schedule functionfromdjango_q.tasksimportscheduleschedule('math.copysign',2,-2,hook='hooks.print_result',schedule_type=Schedule.DAILY)# Or create the object directlyfromdjango_q.modelsimportScheduleSchedule.objects.create(func='math.copysign',hook='hooks.print_result',args='2,-2',schedule_type=Schedule.DAILY)# Run a task every 5 minutes, starting at 6 today# for 2 hoursimportarrowschedule('math.hypot',3,4,schedule_type=Schedule.MINUTES,minutes=5,repeats=24,next_run=arrow.utcnow().replace(hour=18,minute=0))

有关更多信息，请查看Schedules文档。

测试

要运行测试，您需要py.test和pytest-django

待办事项

• 更好的测试和覆盖率
• 减少依赖？

确认

• Django Q的灵感来自于与 Django-RQ和 RQ
• 人类可读散列 HumanHash
• 编辑反馈r/django