modelqueue 2.2.1

ModelQueue 是一个基于Django模型的Apache2许可的任务队列。

例如，在appname/models.py

import modelqueue
from django.db import models

class Task(models.Model):
    data = models.TextField()
    status = modelqueue.StatusField(
        # ^-- Just a models.BigIntegerField
        db_index=True,
        # ^-- Index for faster queries.
        default=modelqueue.Status.waiting,
        # ^-- Waiting state is ready to run.
    )

在appname/management/commands/process_tasks.py中

import modelqueue, time
from django.core.management.base import BaseCommand
from .models import Task

class Command(BaseCommand):

    def handle(self, *args, **options):
        while True:
            task = modelqueue.run(
                Task.objects.all(),
                # ^-- Queryset of models to process.
                'status',
                # ^-- Field name for model queue.
                self.process,
                # ^-- Callable to process model.
            )
            if task is None:
                time.sleep(1)
                # ^-- Bring your own parallelism/concurrency.

    def process(self, task):
        pass  # Process task models.

在appname/admin.py中

class TaskAdmin(admin.ModelAdmin):
    actions = [*modelqueue.admin_actions('status')]
    # ^-- Change task status in admin.
    list_filter = [
        modelqueue.admin_list_filter('status'),
        # ^-- Filter tasks in admin by queue state.
    ]

    def get_changeform_initial_data(self, request):
        # v-- Automatically fill in status field when adding a new task.
        return {'status': int(modelqueue.Status.waiting())}

ModelQueue 是一个危险的项目。它将一个坏主意变得简单而有效。你可能会后悔将数据库用作任务队列，但今天不会是那样！

功能

• 纯Python

• 支持Django的管理界面

• 任务可以重试、中止和取消

• 支持每个任务多次尝试

• 使用线程、多进程或asyncio自行提供并行性

• 性能很重要（在模型中添加单个64位字段）

• 完全文档化

• 100%测试覆盖率

• 多年的生产环境压力测试

• 在Python 3.10上开发

• 与所有Django版本兼容

• 在CPython 3.6、3.7、3.8、3.9、3.10上测试

• 在Linux、Mac OS X和Windows上测试

$ python -m pip install modelqueue

>>> import modelqueue
>>> help(modelqueue)