Google Cloud Tasks 和 Google App Engine Python 3

Posted 2023-02-16

【中文标题】Google Cloud Tasks 和 Google App Engine Python 3

【英文标题】：Google Cloud Tasks & Google App Engine Python 3

【发布时间】：2019-12-30 20:08:12

【问题描述】：

我正在尝试使用Google Cloud Tasks API

在 python2.7 应用引擎标准中，您拥有这个惊人的库 (deferred)，它允许您轻松地将工作人员分配给可以异步完成的多个任务。

所以在 webapp2 处理程序中我可以这样做：

create_csv_file(data):
  #do a bunch of work ...
  return

MyHandler(webapp2.Handler):
  def get(self)
    data = myDB.query()
    deferred.defer(create_csv_file, data)

现在我正在开发新的 Google App Engine Python 3 运行时，而延迟库不适用于 GAE Py3。

谷歌云任务是正确的解决方案/替代品吗？

这就是我现在所处的位置...我已经在互联网上搜索了答案，但我的 Google 功能让我失望了。我找到了一些示例，但它们不是很好，它们看起来好像您应该从 gcloud 控制台或本地创建创建/添加任务，但没有从前端 api 端点添加任务的示例。

ExportCSVFileHandler(Resource):
  def get(self):
    create_task()
    return 

CSVTaskHandler(Resource):
  def(post):
    #do a lot of work creating a csv file
    return

create_task():
    client = tasks.CloudTasksClient(credentials='mycreds')
    project = 'my-project_id'
    location = 'us-east4'
    queue_name = 'csv-worker'

    parent = client.location_path(project, location)

    the_queue = 
        'name': client.queue_path(project, location, queue_name),
        'rate_limits': 
            'max_dispatches_per_second': 1
        ,
        'app_engine_routing_override': 
            'version': 'v2',
            'service': 'task-module'
        
    

    queues = [the_queue]
    task = 
        'app_engine_http_request': 
            'http_method': 'GET',
            'relative_uri': '/create-csv',
            'app_engine_routing': 
                'service': 'worker'
            ,
            'body': str(20).encode()
        
    

    # Use the client to build and send the task.
    response = client.create_task(parent, task)

    print('Created task '.format(response.name))

    # [END taskqueues_using_yaml]
    return response

【参考方案1】：

是的，Cloud Tasks 是 App Engine 任务队列的替代品。可以从任何地方调用 API，例如本地、App Engine、外部服务，甚至 gcloud。这些示例向您展示了如何在本地执行此操作，但您可以轻松地将旧任务队列代码替换为新的 Cloud Tasks 库。

很遗憾，Cloud Tasks 没有延迟库。有多种方法可以解决这个问题。为任务处理程序创建单独的端点并使用 App Engine 路由将任务发送到正确的端点，或将元数据添加到任务正文，以便您的处理程序适当地处理任务请求。

【讨论】：

另外我还会提供关于migrating from Task Queues to Cloud Tasks的文档

你知道 github 等上的任何示例代码显示从 Flask 处理程序（或类似的处理程序）调用的任务 API 吗？

@bscott，谷歌目前没有示例，但这里是我整理的一些示例代码github.com/averikitsch/python-docs-samples/blob/flask-task/…