模块 8：从 App Engine ndb 和任务队列迁移到 Cloud NDB 和 Cloud Tasks

这一系列的 Codelab（自定进度的动手教程）旨在帮助 Google App Engine（标准）开发者通过一系列迁移来指导他们的应用现代化。最重要的步骤是摆脱原始的运行时捆绑服务，因为下一代运行时更加灵活，为用户提供了更多的服务选项。移至新一代运行时，您可以更轻松地与 Google Cloud 产品集成，使用更多支持的服务，并支持当前的语言版本。

此 Codelab 帮助用户从 App Engine 推送任务及其 taskqueue API/库迁移到 Cloud Tasks。如果您的应用没有使用任务队列，您可以使用此 Codelab 作为练习，了解如何将 App Engine 推送任务迁移到 Cloud Tasks。

您将了解如何

从 App Engine taskqueue 迁移到 Cloud Tasks
使用 Cloud Tasks 创建推送任务
从 App Engine ndb 迁移到 Cloud NDB（与模块 2 相同）

所需条件

符合以下条件的 Google Cloud Platform 项目：
- 活跃的 GCP 结算帐号
- 已启用 App Engine
基本 Python 技能
常用 Linux 命令的实践知识
具备开发和部署 App Engine 应用的基础知识
我们建议您在学习本模块（模块 8）之前完成模块 7 Codelab。
有效的模块 7 App Engine 应用

调查问卷

如何使用此 Codelab？

只是阅读

阅读并完成练习

由于我们已在之前的（模块 7）Codelab 中将 App Engine 推送任务添加到示例应用中，因此我们现在可以将其迁移到 Cloud Tasks。本教程的迁移功能包括以下主要步骤：

设置/准备工作
更新配置文件
更新主应用

在开始学习本教程的主要部分之前，让我们设置项目、获取代码，然后部署基准应用，以便我们知道我们从工作代码开始。

1.设置项目

我们建议您重复使用与完成模块 7 Codelab 相同的项目。或者，您可以创建一个全新的项目或重复使用另一个现有项目。确保该项目具有有效的结算帐号，并且已启用 App Engine（应用）。

2.获取基准示例应用

此 Codelab 的前提条件是拥有有效的工作模块 7 示例应用。如果您没有此模块，我们建议您先完成模块 7 教程（上面的链接），然后再进行学习。或者，如果您已经熟悉其内容，可以直接从下面的模块 7 开始。

无论您是使用自己的代码还是我们的代码，我们都能在模块 7 中。本模块 2 Codelab 将会逐步引导您完成每个步骤，完成之后，它应该类似于 FINISH 点的代码（包括从 Python 2 到 3 的可选端口）。

START：模块 7 代码库
FINISH：模块 8 代码库
整个代码库（克隆或下载 ZIP）

模块 7 文件（您或我们的文件）的目录应如下所示：

$ ls
README.md               appengine_config.py     requirements.txt
app.yaml                main.py                 templates

学完模块 7 教程后，您还将有一个包含 Flask 及其依赖项的 lib 文件夹。

3.（重新）部署模块 7 应用

现在，您需要执行剩余的准备工作步骤：

熟悉 gcloud 命令行工具（必要的话）
（重新）将模块 7 代码部署到 App Engine（必要的话）

当您成功执行这些步骤并确认操作有效后，我们将在本教程中进行本教程，从配置文件开始。

`requirements.txt`

模块 7 中的 requirements.txt 仅将 Flask 列为必需的软件包。Cloud NDB 和 Cloud Tasks 具有自己的客户端库，因此，在本步骤中，请将这些软件包添加到 requirements.txt，使其如下所示：

Flask==1.1.2
google-cloud-ndb==1.7.1
google-cloud-tasks==1.5.0

我们建议使用每个库的最新版本；上述版本号是写入本文时 Python 2 的最新版本号。（Python 3 等效软件包可能会拥有较高的版本。）FINISH 代码库文件夹中的代码更新频率更高，并且可能具有较新的版本，不过对于那些通常冻结的 Python 2 库而言，这不太可能是如此。

`app.yaml`

在 libraries 部分的 app.yaml 中引用 grpcio 和 setuptools 内置库：

libraries:
- name: grpcio
  version: 1.0.0
- name: setuptools
  version: 36.6.0

`appengine_config.py`

更新 appengine_config.py 以使用 pkg_resources 将这些内置库与复制的第三方库（如 Flask 和 Google Cloud 客户端库）相关联：

import pkg_resources
from google.appengine.ext import vendor

# Set PATH to your libraries folder.
PATH = 'lib'
# Add libraries installed in the PATH folder.
vendor.add(PATH)
# Add libraries to pkg_resources working set to find the distribution.
pkg_resources.working_set.add_entry(PATH)

只有一个应用文件 main.py，因此此部分中的所有更改只会影响该文件。

更新导入和初始化

我们的应用目前使用内置的 google.appengine.api.taskqueue 和 google.appengine.ext.ndb 库：

之前：

from datetime import datetime
import logging
import time
from flask import Flask, render_template, request
from google.appengine.api import taskqueue
from google.appengine.ext import ndb

将两者替换为 google.cloud.ndb 和 google.cloud.tasks。此外，Cloud Tasks 要求您对任务的载荷进行 JSON 编码，因此也导入 json。完成后，main.py 的 import 部分应如下所示：

之后：

from datetime import datetime
import json
import logging
import time
from flask import Flask, render_template, request
from google.cloud import ndb, tasks

迁移到 Cloud Tasks（和 Cloud NDB）

之前：

def store_visit(remote_addr, user_agent):
    'create new Visit entity in Datastore'
    Visit(visitor='{}: {}'.format(remote_addr, user_agent)).put()

除了在模块 2 中所做的那样，store_visit() 没有变化：向所有 Datastore 访问添加上下文管理器。这种方式会以一种方式创建在 with 语句中新建的 Visit 实体。

之后：

def store_visit(remote_addr, user_agent):
    'create new Visit entity in Datastore'
    with ds_client.context():
        Visit(visitor='{}: {}'.format(remote_addr, user_agent)).put()

目前，Cloud Tasks 要求为您的 Google Cloud 项目启用 App Engine 才能使用它（即使您没有任何 App Engine 代码），否则任务队列将无法正常运行。（如需了解详情，请参阅文档中的此部分。）Cloud Tasks 支持在 App Engine（App Engine“目标”）上运行的任务，但也可以在具有公共 IP 地址的任何 HTTP 端点（HTTP 目标）上运行，例如 Cloud Functions、Cloud Run、GKE、Compute Engine 甚至是本地网络服务器。我们的简单应用使用 App Engine 目标来完成任务。

您需要进行一些设置才能使用 Cloud NDB 和 Cloud Tasks。在 main.py 顶部 Flask 初始化下方，初始化 Cloud NDB 和 Cloud Tasks。还要定义一些常量，以指示推送任务将在何处执行。

app = Flask(__name__)
ds_client = ndb.Client()
ts_client = tasks.CloudTasksClient()

PROJECT_ID = 'PROJECT_ID'  # replace w/your own
REGION = 'REGION'    # replace w/your own
QUEUE_NAME = 'default'     # replace w/your own if desired
QUEUE_PATH = ts_client.queue_path(PROJECT_ID, REGION, QUEUE_NAME)

创建任务队列后，请填写项目的 PROJECT_ID，任务将在其中运行的 REGION（应与 App Engine 区域相同）以及推送队列的名称。App Engine 具有“default”队列，因此我们将使用该名称（但您不必）。

default 默认队列是特殊的，并且在某些情况下会自动创建，其中之一是在使用 App Engine API 时，因此，如果您（重新）使用与模块 7 相同的项目，则 default 队列已经存在。但是，如果您专门为模块 8 创建了一个新项目，则需要手动创建 default 项目。如需详细了解 default 队列，请参阅 queue.yaml 文档。

ts_client.queue_path() 的目的是创建任务队列的“完全限定路径名”(QUEUE_PATH)。此外，还需要使用一种 JSON 结构来指定任务参数：

task = {
    'app_engine_http_request': {
        'relative_uri': '/trim',
        'body': json.dumps({'oldest': oldest}).encode(),
        'headers': {
            'Content-Type': 'application/json',
        },
    }
}

您在上面看到的是什么？

提供任务目标信息：
- 对于 App Engine 目标，将 app_engine_http_request 指定为请求类型，而 relative_uri 是 App Engine 任务处理程序。
- 对于 HTTP 目标，请改用 http_request 和 url。
body：要发送到（推送）任务的 JSON 和 Unicode 字符串编码参数
明确指定 JSON 编码的 Content-Type 标头

请参阅文档，详细了解这里的选项。

设置完毕后，让我们更新 fetch_visits()。上一个教程中的代码如下所示：

之前：

def fetch_visits(limit):
    'get most recent visits and add task to delete older visits'
    data = Visit.query().order(-Visit.timestamp).fetch(limit)
    oldest = time.mktime(data[-1].timestamp.timetuple())
    oldest_str = time.ctime(oldest)
    logging.info('Delete entities older than %s' % oldest_str)
    taskqueue.add(url='/trim', params={'oldest': oldest})
    return (v.to_dict() for v in data), oldest_str

必要的更新：

从 App Engine ndb 切换到 Cloud NDB
显示提取最早访问时间戳的新代码
使用 Cloud Tasks 而不是 App Engine 创建新任务 taskqueue

您的新 fetch_visits() 应如下所示：

之后：

def fetch_visits(limit):
    'get most recent visits and add task to delete older visits'
    with ds_client.context():
        data = Visit.query().order(-Visit.timestamp).fetch(limit)
    oldest = time.mktime(data[-1].timestamp.timetuple())
    oldest_str = time.ctime(oldest)
    logging.info('Delete entities older than %s' % oldest_str)
    task = {
        'app_engine_http_request': {
            'relative_uri': '/trim',
            'body': json.dumps({'oldest': oldest}).encode(),
            'headers': {
                'Content-Type': 'application/json',
            },
        }
    }
    ts_client.create_task(parent=QUEUE_PATH, task=task)
    return (v.to_dict() for v in data), oldest_str

总结代码更新：

切换到 Cloud NDB 意味着将 Datastore 代码移入 with 语句
切换到 Cloud Tasks 意味着使用 ts_client.create_task() 而不是 taskqueue.add()
传入队列的完整路径和 task 载荷（如前所述）

更新（推送）任务处理程序

对（推送）任务处理程序函数几乎需要进行一些更改。

之前：

@app.route('/trim', methods=['POST'])
def trim():
    '(push) task queue handler to delete oldest visits'
    oldest = request.form.get('oldest', type=float)
    keys = Visit.query(
            Visit.timestamp < datetime.fromtimestamp(oldest)
    ).fetch(keys_only=True)
    nkeys = len(keys)
    if nkeys:
        logging.info('Deleting %d entities: %s' % (
                nkeys, ', '.join(str(k.id()) for k in keys)))
        ndb.delete_multi(keys)
    else:
        logging.info('No entities older than: %s' % time.ctime(oldest))
    return ''   # need to return SOME string w/200

唯一需要执行的操作是将所有 Datastore 访问权限都放在上下文管理器 with 语句中，包括查询和删除请求。考虑到这一点，请更新您的 trim() 处理程序，如下所示：

之后：

@app.route('/trim', methods=['POST'])
def trim():
    '(push) task queue handler to delete oldest visits'
    oldest = float(request.get_json().get('oldest'))
    with ds_client.context():
        keys = Visit.query(
                Visit.timestamp < datetime.fromtimestamp(oldest)
        ).fetch(keys_only=True)
        nkeys = len(keys)
        if nkeys:
            logging.info('Deleting %d entities: %s' % (
                    nkeys, ', '.join(str(k.id()) for k in keys)))
            ndb.delete_multi(keys)
        else:
            logging.info('No entities older than: %s' % time.ctime(oldest))
    return ''   # need to return SOME string w/200

此代码和下一个 Codelab 中均不需要更改 templates/index.html。

部署应用

仔细检查您的代码编译后的所有更改，并重新部署。确认应用（仍）正常运行。您应该会看到与模块 7 相同的输出。您只需要在后台重新进行相关操作，所有代码应该都能按预期运行。

如果您在没有进行模块 7 Codelab 的情况下跳转至本教程，则该应用本身不会更改。它会记录对主页 (/) 的所有访问，并且一旦您访问了该网站足够多的时间，它就会看起来像这样，并告诉您它已删除所有早于第十个访问的访问：

模块 7 visitme 应用

这个 Codelab 到此结束。您的代码现在应与模块 8 代码库中的内容相匹配。恭喜完成推送任务迁移最重要的任务！模块 9（下方 Codelab 链接）是可选的，可帮助用户迁移至 Python 3 和 Cloud Datastore。

可选：清理

何不准备好清理，以避免在进入下一个迁移 Codelab 时继续计费？作为现有开发者，您可能已经对 App Engine 的价格信息已经跟上了速度。

可选：停用应用

如果您还没准备好学习下一个教程，请停用应用，以免产生费用。准备好运行下一个 Codelab 后，可以重新启用它。在您的应用被停用的情况下，它不会获取任何流量来产生费用，但是您还需要计费的另一事项是，如果 Datastore 使用量超出免费配额，因此请删除一部分使用量，以使其低于该限制。

另一方面，如果您不打算继续迁移，并想删除所有内容，则可以关停项目。

后续步骤

在本教程中，下一步是模块 9 及其 Codelab 和移植到 Python 3。您可以选择性地进行设置，因为并非所有人都已准备好执行此步骤。Cloud NDB 到 Cloud Datastore 的端口是可选的 - 可选，仅适用于那些将从 NDB 迁出并合并使用 Cloud Datastore 的代码的人员；迁移与模块 3 迁移 Codelab 完全相同。

模块 9 从 Python 2 迁移到 3，以及 Cloud NDB 迁移到 Cloud Datastore
- 移植到 Python 3 的可选迁移模块
- 还包括从 Cloud NDB 迁移到 Cloud Datastore（与模块 3 相同）以及
- 从 Cloud Tasks v1 迁移到 v2 的次要迁移（其客户端库在 Python 2 中冻结）
模块 4：使用 Docker 迁移到 Cloud Run
- 将应用容器化以使用 Docker 在 Cloud Run 上运行
- 通过此迁移，您可以继续使用 Python 2。
模块 5：使用 Cloud Buildpack 迁移到 Cloud Run
- 将应用设置为使用 Cloud Buildpack 在 Cloud Run 上运行
- 您无需了解 Docker、容器或 Dockerfile。
- 要求您的应用已迁移到 Python 3（Buildpack 不支持 Python 2）
模块 6：迁移到 Cloud Firestore
- 迁移到 Cloud Firestore 以使用 Firebase 功能
- Cloud Firestore 支持 Python 2，但此 Codelab 仅在 Python 3 中可用。

App Engine 迁移模块 Codelab 问题/反馈

如果您遇到此 Codelab 的任何问题，请先搜索您的问题，然后再提交。用于搜索新问题和创建新问题的链接：

迁移资源

下表提供了指向模块 7 (START) 和模块 8 (FINISH) 的代码库文件夹的链接。您也可以通过所有 App Engine Codelab 迁移的代码库访问这些代码库，该代码库可以克隆或下载 ZIP 文件。

Codelab	Python 2	Python 3
模块 7	代码	（不适用）
模块 8	代码	（不适用）

App Engine 资源

以下是有关此具体迁移的其他资源：

App Engine ndb 和 Cloud NDB 参考
App Engine taskqueue 和 Cloud Tasks 参考文档
迁移到 Python 3 和 GAE 下一代运行时
常规
- 在 Google Cloud Platform 上使用 Python 应用
- Google Cloud Python 客户端库