一个简化的包,可以在5分钟内将google云任务集成到django应用程序中。
gcloudtasks的Python项目详细描述
一个简化的包,可以在5分钟内将Google云任务集成到django应用程序中。在
文件
快速启动
安装django gcloud任务:
pip install gcloudtasks
将其添加到已安装的应用程序中:
^{pr2}$然后运行迁移命令
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()
历史
- PyPI的第一个版本。在
- 项目
标签: