Google App Engine中TaskQueue无法运行，AssertionError报错求助

在基于Python3的Django项目中使用Google App Engine的TaskQueue功能时，运行触发如下错误：

File "/layers/google.python.pip/pip/lib/python3.8/site-packages/django/core/handlers/exception.py", line 55, in inner
    response = get_response(request)
  File "/layers/google.python.pip/pip/lib/python3.8/site-packages/django/core/handlers/base.py", line 197, in _get_response
    response = wrapped_callback(request, *callback_args, **callback_kwargs)
  File "/layers/google.python.pip/pip/lib/python3.8/site-packages/django/views/generic/base.py", line 103, in view
    return self.dispatch(request, *args, **kwargs)
  File "/layers/google.python.pip/pip/lib/python3.8/site-packages/django/views/generic/base.py", line 142, in dispatch
    return handler(request, *args, **kwargs)
  File "/workspace/apps/account/views.py", line 235, in post
    taskqueue.add(queue_name='create-waybill', url = reverse('_api:createWaybill') ,  params = {'key':url_key})

  File "/layers/google.python.pip/pip/lib/python3.8/site-packages/google/appengine/api/taskqueue/taskqueue.py", line 575, in create_rpc
    return apiproxy_stub_map.UserRPC('taskqueue', deadline, callback)
  File "/layers/google.python.pip/pip/lib/python3.8/site-packages/google/appengine/api/apiproxy_stub_map.py", line 444, in __init__
    self.__rpc = CreateRPC(service, stubmap)
  File "/layers/google.python.pip/pip/lib/python3.8/site-packages/google/appengine/api/apiproxy_stub_map.py", line 69, in CreateRPC
    assert stub, 'No api proxy found for service "taskqueue"'
AssertionError: No api proxy found for service "taskqueue"

我的TaskQueue相关代码：

from google.appengine.api import taskqueue

taskqueue.Task(name='create-waybill', url='/_api/createWaybill',params = {'key':url_key}).add()

对应的URL路由配置：

path ('createWaybill', csrf_exempt(views.CreateWaybill.as_view()), name='createWaybill'),

解决办法 #

1. 确认运行环境兼容性 #

google.appengine.api.taskqueue是GAE标准环境专属的旧API，不支持灵活环境：

• 本地开发：必须使用GAE官方的dev_appserver.py启动项目，不能直接用Django的runserver命令。
• 部署到GAE灵活环境：放弃旧TaskQueue API，改用Cloud Tasks API（Google官方推荐的替代方案）。

2. 标准环境下的依赖配置 #

如果使用GAE标准环境，需确保：

• 安装必要依赖：

  pip install appengine-python-standard
  

• app.yaml中配置正确的运行时：

  runtime: python38
  entrypoint: gunicorn -b :$PORT your_project.wsgi
  

3. 切换到Cloud Tasks API的代码示例 #

若使用灵活环境或想升级到最新API，代码调整如下：

from google.cloud import tasks_v2
import os

# 初始化客户端
client = tasks_v2.CloudTasksClient()
project_id = os.getenv('GOOGLE_CLOUD_PROJECT')
location = 'us-central1'  # 替换为你的队列所在区域
queue_name = 'create-waybill'

# 构建队列父路径
parent = client.queue_path(project_id, location, queue_name)

# 构建任务
task = {
    'http_request': {
        'http_method': tasks_v2.HttpMethod.POST,
        # 生产环境用GAE服务域名，本地调试可替换为localhost地址
        'url': f'https://{os.getenv("GAE_SERVICE")}.appspot.com/_api/createWaybill',
        'body': f'key={url_key}'.encode(),
        'headers': {'Content-Type': 'application/x-www-form-urlencoded'}
    }
}

# 创建任务
response = client.create_task(request={"parent": parent, "task": task})

4. 本地调试Cloud Tasks #

本地开发时可使用Cloud Tasks模拟器：

1. 启动模拟器：

   gcloud beta emulators tasks start
   

2. 设置环境变量：

   export CLOUDSDK_CORE_PROJECT=你的项目ID
   export TASKS_EMULATOR_HOST=localhost:8081
   

3. 启动Django项目即可进行调试。

内容的提问来源于stack exchange，提问作者Jaskaran Singh