迁移 Python 2 App Engine Cloud NDB 和从 Cloud Tasks 应用迁移到 Python 3 和 Cloud Datastore（模块 9）

1. 概览

无服务器迁移站系列 Codelab（自定进度的实操教程）和相关视频旨在指导 Google Cloud 无服务器开发者完成一次或多次迁移（主要是从旧服务迁移），从而实现应用的现代化改造。这样做可以提高应用的可移植性，为您提供更多选择和灵活性，让您能够集成并访问更广泛的 Cloud 产品，并更轻松地升级到较新版本。虽然本系列最初主要面向的是最早接触 Cloud 的用户（主要是 App Engine（标准环境）开发者），但涵盖的范围非常广泛，涵盖了 Cloud Functions 和 Cloud Run 等其他无服务器平台，或其他无服务器平台（如适用）。

此 Codelab 的目的是将第 8 单元的示例应用移植到 Python 3，并将 Datastore（Datastore 模式下的 Cloud Firestore）的访问权限从使用 Cloud NDB 切换为使用原生 Cloud Datastore 客户端库，并升级到最新版本的 Cloud Tasks 客户端库。

我们在模块 7 中添加了任务队列用于推送任务，然后在模块 8 中将其迁移到了 Cloud Tasks。在第 9 单元中，我们将继续介绍 Python 3 和 Cloud Datastore。使用任务队列处理“拉取”任务的用户将迁移到 Cloud Pub/Sub，并应参阅模块 18-19。

未使用任务队列？

如果您的应用不使用 App Engine 任务队列服务，请跳过模块 7-9（推送任务）和 18-19（拉取任务），或完成这些 Codelab 作为练习，以熟悉任务队列迁移。

在接下来的实验中

• 将模块 8 示例应用移植到 Python 3
• 将 Datastore 访问权限从 Cloud NDB 切换到 Cloud Datastore 客户端库
• 升级到最新的 Cloud Tasks 客户端库版本

所需条件

• 具有有效的 GCP 结算账号的 Google Cloud Platform 项目
• 基本 Python 技能
• 常用 Linux 命令的实践知识
• 具备开发和部署 App Engine 应用的基础知识
• 具备可正常运行的模块 8 App Engine 应用：完成第 8 单元 Codelab（推荐），或从代码库中复制 Module 8 应用

调查问卷

您打算如何使用本教程？

<ph type="x-smartling-placeholder"></ph> 仅仔细阅读 阅读并完成练习

您如何评价使用 Python 的体验？

新手水平 中等水平 熟练水平

您如何评价自己在使用 Google Cloud 服务方面的经验水平？

<ph type="x-smartling-placeholder"></ph> 新手 中级 熟练

2. 背景

模块 7 演示了如何在 Python 2 Flask App Engine 应用中使用 App Engine 任务队列推送任务。在单元 8 中，您要将该应用从任务队列迁移到 Cloud Tasks。在第 9 单元中，您将继续学习，将该应用移植到 Python 3，并将 Datastore 访问权限从使用 Cloud NDB 切换到原生 Cloud Datastore 客户端库。

由于 Cloud NDB 同时适用于 Python 2 和 3，因此它足以满足 App Engine 用户将应用从 Python 2 移植到 3 的需求。另外，将客户端库迁移到 Cloud Datastore 是完全可选操作，Cloud NDB 是专为 Python 2 App Engine 开发者打造的，用作 Python 3 迁移工具，因此如果您还没有使用 Cloud Datastore 客户端库的代码，则无需考虑此迁移。

最后，Cloud Tasks 客户端库的开发只在 Python 3 中继续，因此我们“迁移”从最终的 Python 2 版本之一升级到了最新的 Python 3 版本。幸运的是，与 Python 2 相比没有重大更改，这意味着您在此无需执行任何其他操作。

本教程包含以下步骤：

1. 设置/准备工作
2. 更新配置
3. 修改应用代码

3. 设置/准备工作

本节介绍如何执行以下操作：

1. 设置 Cloud 项目
2. 获取基准示例应用
3. （重新）部署并验证基准应用

以下步骤可确保您从有效的代码入手，并且已准备好迁移到 Cloud 服务。

1. 设置项目

如果您已完成第 8 单元 Codelab，请重复使用同一项目（和代码）。或者，创建一个全新的项目或重复使用其他现有项目。确保项目具有有效的结算账号和已启用的 App Engine 应用。在本 Codelab 中，您需要找到自己的项目 ID，以便在遇到 PROJECT_ID 变量时使用该 ID。

2. 获取基准示例应用

前提条件之一是具备有效的第 8 单元的 App Engine 应用：完成第 8 单元 Codelab（推荐），或从代码库中复制第 8 单元应用。无论您使用的是自己的代码还是我们的代码，我们都将从模块 8 代码（“START”）着手。此 Codelab 将逐步引导您完成迁移，最后使用与模块 9 代码库文件夹（“FINISH”）中的代码类似的代码。

• START：模块 8 代码库
• 完成：模块 9 代码库
• 整个代码库（克隆或下载 ZIP 文件）

无论您使用哪种模块 7 应用，该文件夹都应如下所示，可能还包含 lib 文件夹：

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

3. （重新）部署并验证基准应用

执行以下步骤来部署模块 8 应用：

1. 删除 lib 文件夹（如有），然后运行 pip install -t lib -r requirements.txt 以重新填充 lib。如果您的开发机器上同时安装了 Python 2 和 Python 3，则可能需要改用 pip2。
2. 确保您已安装并初始化 gcloud 命令行工具，并查看了其使用情况。
3. （可选）如果您不想在发出的每个 gcloud 命令中输入 PROJECT_ID，请使用 gcloud config set project PROJECT_ID 设置您的 Cloud 项目。
4. 使用 gcloud app deploy 部署示例应用
5. 确认应用按预期运行。如果您已完成第 8 单元 Codelab，该应用会显示最常访问您网站的用户以及最近的访问次数（如下图所示）。底部会显示将被删除的旧任务。

4. 更新配置

requirements.txt

新的 requirements.txt 与模块 8 的几乎相同，只有一项重大更改：将 google-cloud-ndb 替换为 google-cloud-datastore。进行此项更改，使 requirements.txt 文件如下所示：

flask
google-cloud-datastore
google-cloud-tasks

此 requirements.txt 文件不含任何版本号，这意味着已选择最新版本。如果出现任何不兼容的情况，使用版本号锁定应用的有效版本是标准做法。

app.yaml

第二代 App Engine 运行时不支持内置的第三方库（如 2.x），也不支持复制非内置库。第三方软件包的唯一要求是在 requirements.txt 中列出它们。因此，可以删除 app.yaml 的整个 libraries 部分。

另一项更新是 Python 3 运行时需要使用能够自行路由的 Web 框架。因此，必须将所有脚本处理程序都改为 auto。不过，由于所有路由都必须更改为 auto，并且此示例应用不提供静态文件，因此任何处理程序都无关紧要，因此请一并移除整个 handlers 部分。

在 app.yaml 中，唯一需要的操作是将运行时设置为受支持的 Python 3 版本（例如 3.10）。请进行此项更改，使新的缩写 app.yaml 仅包含下面这行代码：

runtime: python310

删除 appengine_config.py 和 lib

新一代 App Engine 运行时改进第三方软件包的使用：

• 内置库是指经过 Google 审查并在 App Engine 服务器上提供的库，这可能是因为它们包含不允许开发者部署到云端的 C/C++ 代码 - 这些库在第 2 代运行时中已不再提供。
• 第 2 代运行时不再需要复制非内置库（有时称为“vendoring”或“自捆绑”）。相反，它们应列在 requirements.txt 中，构建系统在部署时会自动代您进行安装。

由于对第三方软件包管理进行了上述更改，因此既不需要 appengine_config.py 文件和 lib 文件夹，也不需要删除它们。在第 2 代运行时中，App Engine 会自动安装 requirements.txt 中列出的第三方软件包。总结：

1. 不得自行捆绑或复制第三方库；在requirements.txt中列出
2. 没有 pip install 到 lib 文件夹中，即无 lib 文件夹期限
3. app.yaml 中未列出内置的第三方库（因此没有 libraries 部分）；在requirements.txt中列出
4. 没有可从您的应用引用的第三方库意味着没有 appengine_config.py 文件

在 requirements.txt 中列出所有需要的第三方库是开发者的唯一要求。

5. 更新应用文件

只有一个应用文件 main.py，因此本部分中的所有更改都只会影响该文件。下面是“差异”部分图示：将现有代码重构到新应用中需要进行的整体更改。读者不需要逐行阅读代码，因为代码只是为了简要直观地了解此次重构所需要的内容（但如果需要，可随时在新标签页中打开或下载并放大）。

更新导入和初始化

第 8 单元的 main.py 中的导入部分使用 Cloud NDB 和 Cloud Tasks；它应如下所示：

之前：

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

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

在第二代运行时（例如 Python 3）中，日志记录进行了简化和增强：

• 如需获得全面的日志记录体验，请使用 Cloud Logging
• 对于简单的日志记录，只需通过 print() 发送到 stdout（或 stderr）
• 无需使用 Python logging 模块（因此请将其移除）

因此，请删除 logging 的导入，并将 google.cloud.ndb 替换为 google.cloud.datastore。同样，将 ds_client 更改为指向 Datastore 客户端而不是 NDB 客户端。完成上述更改后，新应用的顶部现在如下所示：

之后：

from datetime import datetime
import json
import time
from flask import Flask, render_template, request
import google.auth
from google.cloud import datastore, tasks

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

迁移到 Cloud Datastore

现在，是时候用 Datastore 替换 NDB 客户端库了。App Engine NDB 和 Cloud NDB 都需要数据模型（类）；对于此应用，Visit。store_visit() 函数在所有其他迁移模块中的工作方式相同：它通过创建新的 Visit 记录并保存访问客户端的 IP 地址和用户代理（浏览器类型）来注册访问。

之前：

class Visit(ndb.Model):
    'Visit entity registers visitor IP address & timestamp'
    visitor   = ndb.StringProperty()
    timestamp = ndb.DateTimeProperty(auto_now_add=True)

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 Datastore 不使用数据模型类，因此请删除该类。此外，在创建记录时，Cloud Datastore 不会自动创建时间戳，您需要手动创建，可通过 datetime.now() 调用来完成。

如果没有数据类，修改后的 store_visit() 应如下所示：

之后：

def store_visit(remote_addr, user_agent):
    'create new Visit entity in Datastore'
    entity = datastore.Entity(key=ds_client.key('Visit'))
    entity.update({
        'timestamp': datetime.now(),
        'visitor': '{}: {}'.format(remote_addr, user_agent),
    })
    ds_client.put(entity)

关键函数是 fetch_visits()。它不仅会对最新 Visit 执行原始查询，还会获取显示的最后一个 Visit 的时间戳，并创建调用 /trim（因此是 trim()）的推送任务来批量删除旧的 Visit。以下是使用 Cloud NDB 的示例：

之前：

def fetch_visits(limit):
    'get most recent visits & 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

主要变更：

1. 将 Cloud NDB 查询替换为 Cloud Datastore 等效查询；查询样式略有不同
2. Datastore 不需要使用上下文管理器，也不会像 Cloud NDB 那样要求您（使用 to_dict()）提取其数据。
3. 将日志记录调用替换为 print()

完成这些更改后，fetch_visits() 将如下所示：

之后：

def fetch_visits(limit):
    'get most recent visits & add task to delete older visits'
    query = ds_client.query(kind='Visit')
    query.order = ['-timestamp']
    visits = list(query.fetch(limit=limit))
    oldest = time.mktime(visits[-1]['timestamp'].timetuple())
    oldest_str = time.ctime(oldest)
    print('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 visits, oldest_str

通常只需这样操作即可。遗憾的是，有一个严重问题。

（可能）创建新的（推送）队列

在第 7 单元中，我们在现有的第 1 单元应用中添加了 App Engine taskqueue 的使用。将推送任务作为旧版 App Engine 功能的一个主要好处是，“默认”的队列。当该应用在模块 8 中迁移到 Cloud Tasks 时，该默认队列已经存在，因此我们仍然无需为此担心。这一点在单元 9 中有所改变。

需要考虑的一个关键方面是，新的 App Engine 应用不再使用 App Engine 服务，因此，您不能再假设 App Engine 自动在其他产品 (Cloud Tasks) 中自动创建任务队列。如上所示，在 fetch_visits() 中创建任务（针对不存在的队列）将失败。需要一个新函数来检查 "default" 队列是否存在，如果不存在，则创建一个。

调用此函数 _create_queue_if()，并将其添加到您的应用的 fetch_visits() 上方，因为这是调用它的位置。要添加的函数的正文：

def _create_queue_if():
    'app-internal function creating default queue if it does not exist'
    try:
        ts_client.get_queue(name=QUEUE_PATH)
    except Exception as e:
        if 'does not exist' in str(e):
            ts_client.create_queue(parent=PATH_PREFIX,
                    queue={'name': QUEUE_PATH})
    return True

Cloud Tasks create_queue() 函数需要队列的完整路径名，但队列名称除外。为简单起见，请创建另一个表示 QUEUE_PATH 减去队列名称 (QUEUE_PATH.rsplit('/', 2)[0]) 的变量 PATH_PREFIX。在顶部附近添加其定义，使包含所有常量赋值的代码块如下所示：

_, PROJECT_ID = google.auth.default()
REGION_ID = 'REGION_ID'    # replace w/your own
QUEUE_NAME = 'default'     # replace w/your own
QUEUE_PATH = ts_client.queue_path(PROJECT_ID, REGION_ID, QUEUE_NAME)
PATH_PREFIX = QUEUE_PATH.rsplit('/', 2)[0]

现在，将 fetch_visits() 中的最后一行修改为使用 _create_queue_if()，首先根据需要创建队列，然后创建任务：

    if _create_queue_if():
        ts_client.create_task(parent=QUEUE_PATH, task=task)
    return visits, oldest_str

现在，_create_queue_if() 和 fetch_visits() 汇总后应如下所示：

def _create_queue_if():
    'app-internal function creating default queue if it does not exist'
    try:
        ts_client.get_queue(name=QUEUE_PATH)
    except Exception as e:
        if 'does not exist' in str(e):
            ts_client.create_queue(parent=PATH_PREFIX,
                    queue={'name': QUEUE_PATH})
    return True

def fetch_visits(limit):
    'get most recent visits & add task to delete older visits'
    query = ds_client.query(kind='Visit')
    query.order = ['-timestamp']
    visits = list(query.fetch(limit=limit))
    oldest = time.mktime(visits[-1]['timestamp'].timetuple())
    oldest_str = time.ctime(oldest)
    print('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',
            },
        }
    }
    if _create_queue_if():
        ts_client.create_task(parent=QUEUE_PATH, task=task)
    return visits, oldest_str

除了必须添加这些额外的代码之外，第 8 单元中的 Cloud Tasks 代码的其余部分基本保持不变。最后需要查看的代码是任务处理程序。

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

在任务处理程序 trim() 中，Cloud NDB 代码会查询早于显示的最早的访问时间。它使用仅限于键的查询来加快速度。如果只需要访问 ID，为什么还要获取所有数据？获得所有访问 ID 后，使用 Cloud NDB 的 delete_multi() 函数批量删除所有访问 ID。

之前：

@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

与 fetch_visits() 一样，大部分更改涉及将 Cloud NDB 代码换成 Cloud Datastore、调整查询样式、不再使用其上下文管理器，以及将日志记录调用更改为 print()。

之后：

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

主应用处理程序 root() 没有更改。

移植到 Python 3

此示例应用可同时在 Python 2 和 3 上运行。本教程的相关部分已介绍了任何特定于 Python 3 的更改。无需执行其他步骤，也无需使用兼容性库。

Cloud Tasks 更新

支持 Python 2 的最终 Cloud Tasks 客户端库版本是 1.5.0。在撰写本文时，适用于 Python 3 的客户端库的最新版本与该版本完全兼容，因此无需进一步更新。

HTML 模板更新

也无需对 HTML 模板文件 templates/index.html 进行任何更改，因此这样就涵盖生成模块 9 应用所需的所有更改。

6. 摘要/清理

部署并验证应用

完成代码更新（主要是移植到 Python 3）后，使用 gcloud app deploy 部署您的应用。除了您已将数据库访问权限移至 Cloud Datastore 客户端库并升级到 Python 3 以外，输出应该与模块 7 和模块 8 应用中的应用相同：

此步骤已完成 Codelab。不妨将您的代码与“Module 9”文件夹中的代码进行比较。恭喜！

清理

常规

如果您目前已完成，我们建议您停用 App Engine 应用，以免产生费用。不过，如果您希望测试或实验更多内容，App Engine 平台有免费配额，因此只要您不超过该使用量水平，您就不必支付费用。这只是计算费用，但相关 App Engine 服务可能也会产生费用，因此请查看其价格页面了解详情。如果此迁移涉及其他 Cloud 服务，这些服务单独计费。无论是哪种情况（如适用），请参阅“此 Codelab 的具体说明”部分。

为了全面披露，部署到像 App Engine 这样的 Google Cloud 无服务器计算平台会产生少量构建和存储费用。Cloud Build 和 Cloud Storage 都有自己的免费配额。该映像的存储会占用部分配额。但是，如果您居住的区域没有这样的免费层级，请留意您的存储空间用量，以最大限度地降低潜在费用。特定的 Cloud Storage“文件夹”您应检查以下内容：

• console.cloud.google.com/storage/browser/LOC.artifacts.PROJECT_ID.appspot.com/containers/images
• console.cloud.google.com/storage/browser/staging.PROJECT_ID.appspot.com
• 上述存储空间链接取决于您的PROJECT_ID和 *LOC*格式，例如“us”。

另一方面，如果您不打算继续学习此应用或其他相关的迁移 Codelab，而是想彻底删除所有内容，请关停项目。

此 Codelab 的具体内容

下列服务是此 Codelab 独有的服务。有关详情，请参阅各个产品的文档：

• Cloud Tasks 提供免费层级；如需了解详情，请参阅其价格页面。
• App Engine Datastore 服务由 Cloud Datastore（Datastore 模式的 Cloud Firestore）提供，该服务也有一个免费层级；如需了解详情，请参阅其价格页面。

后续步骤

从 App Engine 任务队列推送任务到 Cloud Tasks 的迁移到此结束。此外，第 3 单元还单独介绍了从 Cloud NDB 到 Cloud Datastore 的可选迁移（无任务队列或 Cloud Tasks）。除第 3 单元外，还有另外一些迁移模块侧重于摆脱 App Engine 旧版捆绑式服务，包括：

• 第 2 单元：从 App Engine NDB 迁移到 Cloud NDB
• 第 3 单元：从 Cloud NDB 迁移到 Cloud Datastore
• 模块 12-13：从 App Engine Memcache 迁移到 Cloud Memorystore
• 第 15-16 单元：从 App Engine Blob 存储区迁移到 Cloud Storage
• 模块 18-19：App Engine 任务队列（拉取任务）添加到 Cloud Pub/Sub

App Engine 不再是 Google Cloud 中唯一的无服务器平台。如果您有一个小型 App Engine 应用或功能有限的应用，并希望将其转换为独立的微服务，或者您希望将单体式应用拆分为多个可重复使用的组件，那么这些都是考虑迁移到 Cloud Functions 的充分理由。如果容器化已成为应用开发工作流的一部分，特别是当它由 CI/CD（持续集成/持续交付或部署）流水线组成时，请考虑迁移到 Cloud Run。以下模块介绍了这些场景：

• 从 App Engine 迁移到 Cloud Functions：请参阅单元 11
• 从 App Engine 迁移到 Cloud Run：请参阅第 4 单元，了解如何使用 Docker 将应用容器化；请参阅第 5 单元，在不具备容器、Docker 知识或 Dockerfile 的情况下实现容器化

您可以自行选择是否切换到其他无服务器平台，我们建议您先考虑最适合您的应用和用例的方案，然后再做任何更改。

无论您接下来考虑使用哪种迁移模块，都可以通过其开源代码库访问所有 Serverless Migration Station 内容（Codelab、视频、源代码 [如果有]）。代码库的 README 还提供了有关应考虑哪些迁移以及任何相关“顺序”的指南。迁移模块

7. 其他资源

Codelab 问题/反馈

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

• 搜索问题
• 提交新问题/反馈

迁移时可参考的资源

下表列出了模块 8（START）和模块 9 (FINISH) 对应的代码库文件夹的链接。您还可以从所有 App Engine Codelab 迁移的代码库访问这些库，您可以克隆或下载 ZIP 文件。

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

在线资源

以下是可能与本教程相关的在线资源：

App Engine

• App Engine 文档
• Python 2 App Engine（标准环境）运行时
• Python 3 App Engine（标准环境）运行时
• Python 2 与3 个 App Engine（标准环境）运行时
• Python 2 到 3 App Engine（标准环境）迁移指南
• App Engine 价格和配额信息

Cloud NDB

• Google Cloud NDB 文档
• Google Cloud NDB 代码库
• Cloud Datastore 价格信息

Cloud Datastore

• Google Cloud Datastore 文档
• Google Cloud Datastore 代码库
• Cloud Datastore 价格信息

Cloud Tasks

• Google Cloud Tasks 文档
• Google Cloud Tasks 代码库
• Cloud Tasks 价格信息

其他 Cloud 信息

• 在 Google Cloud Platform 上使用 Python 应用
• Google Cloud Python 客户端库
• Google Cloud“始终免费”计划层级
• Google Cloud SDK（gcloud 命令行工具）
• 所有 Google Cloud 文档

许可

此作品已获得 Creative Commons Attribution 2.0 通用许可授权。