gcloudtasks 0.0.12

文件

快速启动

安装django gcloud任务：

pip install gcloudtasks

将其添加到已安装的应用程序中：

然后运行迁移命令

python manage.py migrate

在将要处理任务的实例上，添加django gcloud任务的URL模式：

urlpatterns = [
    ...
    path('__tasks/', include('gcloudtasks.urls')),
    ...
]

您需要配置一些设置。可用设置包括：

• GTASK_PROJECT_ID：（必选）这是您从Google控制台获得的项目ID。在
• GTASK_PROJECT_LOCATION：（必选）您的项目位置，Google控制台的默认位置通常是us-central1，但我仍然建议您确认您的位置。在
• GTASK_CREDENTIAL_FILE：绝对文件到你的Google帐户凭据json文件。如果GOOGLE_APPLICATION_CREDENTIALS设置为环境变量，则不需要此项。在
• GTASK_TARGET_WORKER_HOST：（必选）将处理任务的工作机实例的绝对URL。（例如https://example.com）
• GTASK_SEND_TO_REMOTE：默认为True，表示任务是否应该转发到googletasks远程队列。您可能希望在开发环境中将其更改为False。在
• GTASK_MODEL_CLASS：默认为'gcloudtasks.Task’. 如果您打算使用自定义模型来进行任务管理，则需要更改此设置。在
• GTASK_HANDLER_NAMESPACE：默认为空。仅当您将命名空间添加到URL模式条目时才需要。否则，没有必要。在

信号

要使用这些信号，只需创建如下接收器：

from gcloudtasks.tasks import task_created_signal

@receiver(task_created_signal)
def some_function(sender, instance, created, **kwargs):
    pass

扩展任务模型

创建自定义任务模型：

假设这个模型文件是campaigns/models.py

from gcloudtasks.models import Task

class CustomTask(Task):
    user = models.ForeignKey(settings.AUTH_USER_MODEL, on_delete=models.CASCADE)
    ...
    ...

然后进来设置.py, 关于文件名，我们可以：

GTASK_MODEL_CLASS = 'campaigns.CustomTask'

功能

Create Task：要创建任务，只需用@task装饰函数。 这个修饰符接受两个参数queue和{}。只有queue 是强制性的。queue是要运行任务的任务队列的名称。 如果未提供，则自动生成名称。 如果您还没有队列，有一个函数可以让您检查并创建它 不存在。在

# To create get or create queue

from gcloudtasks.queues import get_or_create_queue

get_or_create_queue(queue_name)

要真正创建任务，您可以创建一个专门的文件并将其命名为tasks.py或任何您想要添加的文件 你的任务在那里发挥作用。在

from gcloudtasks.executor import task


@task(queue='my-default-queue')
def sum_numbers(a, b):
    # do something in the function
    pass

然后，您可以在views.py或任何地方执行类似的操作来执行此任务

from .tasks import sum_numbers

class SomeView(FormView):
    ....

    def form_valid(self, form):
        ...
        sum_numbers(a=10, b=20).run()

您会注意到这些参数是作为关键字参数传递的。这是强制性的。 如果您想安排这个任务，run接受两个可能的非必需参数； delay和{}。delay应该是datetime.timedelta实例 而scheduled_time应该是一个时区感知的datetime实例。在

将sum_numbers函数安排在2天后。我会的

sum_numbers(a=10, b=20).run(delay=timedelta(days=2))

特点

• 任务管理通过django管理。（通过管理员删除任务时，如果任务尚未处理，则会从队列中删除这些任务）。在
• 可扩展任务模型。在
• 任务创建模型信号。在
• 对库的扩展支持，如django-tenant-schemas。在

扩展支持

Django租户模式在编写库时被考虑在内。据了解 您可能希望将任务路由到工作机实例或任何服务器上的特定域 处理任务，因此可以通过 只需将host参数传递到run函数中，如下所示。在

sum_numbers(a=10, b=20).run(host='https://somedomain.example.com')

哥查斯

• 只向任务函数传递JSON可序列化对象。如果您需要一个模型，那么传递id并在fu中检索模型行动。e、 g

# Inside your tasks.py or wherever your task functions reside.
@task(queue='my-default-queue')
def mark_as_done(todo_id):
    todo = Todo.objects.get(pk=todo_id)
    ...


# Where the task is being called
mark_as_done(todo_id=todo.id).run()

学分

用于呈现此包的工具：

• Cookiecutter
• cookiecutter-djangopackage