如何在 Flask 应用中使用 App Engine 任务队列（推送任务）（模块 7）

1. 概览

无服务器迁移站系列 Codelab（自定进度的动手教程）和相关视频旨在帮助 Google Cloud 无服务器开发者通过一次或多次迁移（主要是从旧版服务迁移）来指导他们的应用现代化。这样做可让您的应用更易于移植，并为您提供更多选择和灵活性，使您能够与更多 Cloud 产品集成并访问这些产品，还能更轻松地升级到新的语言版本。虽然最初侧重于最早的 Cloud 用户（主要是 App Engine [标准环境] 开发者），但本系列文章的范围足够广，可涵盖其他无服务器平台（例如 Cloud Functions 和 Cloud Run），或者其他平台（如果适用）。

本 Codelab 将介绍如何在模块 1 Codelab 中的示例应用中使用 App Engine 任务队列推送任务。模块 7 博文和视频可作为本教程的补充，简要介绍本教程中的内容。

在本模块中，我们将添加对推送任务的使用，然后在模块 8 中将该使用迁移到 Cloud Tasks，并在模块 9 中进一步迁移到 Python 3 和 Cloud Datastore。如果使用 Task Queues 处理拉取任务，则需要迁移到 Cloud Pub/Sub，并应改为参阅模块 18-19。

在接下来的实验中

使用 App Engine 任务队列 API/捆绑服务
向基本的 Python 2 Flask App Engine NDB 应用添加推送任务使用情况

所需条件

具有有效的 GCP 结算账号的 Google Cloud 项目
基本 Python 技能
常用 Linux 命令的实践知识
具备开发和部署 App Engine 应用的基础知识
有效的模块 1 App Engine 应用（完成其 Codelab [推荐] 或从代码库复制应用）

调查问卷

您将如何使用本教程？

仅阅读教程内容

阅读并完成练习

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

新手水平

中等水平

熟练水平

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

新手水平

中等水平

熟练水平

2. 背景

App Engine Task Queue 同时支持推送任务和拉取任务。为了提高应用的可移植性，Google Cloud 团队建议从任务队列等旧版捆绑服务迁移到其他 Cloud 独立服务或第三方等效服务。

任务队列推送任务用户应迁移到 Cloud Tasks。
任务队列拉取任务的用户应迁移到 Cloud Pub/Sub。

拉取任务迁移在迁移模块 18-19 中介绍，而模块 7-9 侧重于推送任务迁移。为了从 App Engine 任务队列推送任务进行迁移，请将该任务队列的用法添加到模块 1 Codelab 中生成的现有 Flask 和 App Engine NDB 应用。在该应用中，新的网页浏览量会注册新的访问，并向用户显示最近的访问。由于旧的访问记录永远不会再次显示，并且会占用数据存储区中的空间，因此我们将创建一个推送任务来自动删除最旧的访问记录。在模块 8 中，我们将把该应用从任务队列迁移到 Cloud Tasks。

本教程包含以下步骤：

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

3. 设置/准备工作

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

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

这些步骤可确保您从正常运行的代码开始。

1. 设置项目

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

2. 获取基准示例应用

此 Codelab 的前提条件之一是拥有一个有效的工作模块 1 App Engine 应用：完成模块 1 Codelab（推荐）或从代码库中复制模块 1 应用。无论您是使用自己的代码还是我们的代码，我们都能在模块 1 中“开始”。本 Codelab 将引导您完成每个步骤，最终获得类似于模块 7 代码库文件夹“FINISH”中的代码。

START：模块 1 文件夹 (Python 2)
FINISH：模块 7 文件夹 (Python 2)
整个代码库（用于克隆或下载 ZIP 文件）

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

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

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

执行以下步骤以（重新）部署模块 1 应用：

删除 lib 文件夹（如果有），然后运行 pip install -t lib -r requirements.txt 以重新填充 lib。如果您同时安装了 Python 2 和 3，可能需要改用 pip2 命令。
确保您已安装并初始化 gcloud 命令行工具，并已查看其用法。
如果您不想在每次发出 gcloud 命令时都输入 PROJECT_ID，请使用 gcloud config set project PROJECT_ID 设置您的云项目。
使用 gcloud app deploy 部署示例应用
确认模块 1 应用运行正常，可显示最近的访问记录（如下图所示）

4. 更新配置

无需对标准 App Engine 配置文件（app.yaml、requirements.txt、appengine_config.py）进行任何更改。

5. 修改应用文件

主要应用文件是 main.py，此部分中的所有更新均与该文件有关。此外，网页模板 templates/index.html 也进行了小幅更新。本部分中要实现的更改如下：

更新了导入
添加推送任务
添加任务处理程序
更新网站模板

1. 更新了导入

导入 google.appengine.api.taskqueue 会引入任务队列功能。还需要一些 Python 标准库软件包：

由于我们要添加一项任务来删除最旧的访问记录，因此应用需要处理时间戳，这意味着需要使用 time 和 datetime。
为了记录与任务执行相关的有用信息，我们需要 logging。

添加所有这些导入后，您的代码在进行这些更改前后的样子如下所示：

之前：

from flask import Flask, render_template, request
from google.appengine.ext import 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

2. 添加推送任务（整理任务数据，将新任务加入队列）

推送队列文档中指出：“要处理任务，您必须将其添加到推送队列。App Engine 提供了一个名为 default 的默认推送队列，该队列已预先配置默认设置，可直接使用。如果您喜欢，可以将所有任务都添加到默认队列，而不必创建和配置其他队列。”为了简洁起见，此 Codelab 使用了 default 队列。如需详细了解如何定义具有相同或不同特征的自有推送队列，请参阅创建推送队列文档。

此 Codelab 的主要目标是添加一个任务（到 default 推送队列），该任务负责从 Cloud Datastore 中删除不再显示的旧访问记录。基准应用通过创建新的 Visit 实体来注册每次访问（向 / 发出的 GET 请求），然后提取并显示最近的访问。最旧的访问记录将永远不会再次显示或使用，因此推送任务会删除所有比显示的最旧访问记录更旧的访问记录。为此，应用的行为需要稍作更改：

在查询最近的访问记录时，不要立即返回这些访问记录，而是修改应用以保存最近一次 Visit（即显示的最旧访问记录）的时间戳，删除比此时间戳更旧的所有访问记录是安全的。
创建一个推送任务，将此时间戳作为其载荷，并通过 HTTP POST 将其定向到可访问的 /trim 任务处理程序。具体来说，使用标准 Python 实用程序转换 Datastore 时间戳并将其（以浮点数形式）发送到任务，同时记录该时间戳（以字符串形式）并返回该字符串作为标记值以显示给用户。

所有这些都发生在 fetch_visits() 中，以下是进行这些更新前后的外观：

之前：

def fetch_visits(limit):
    return (v.to_dict() for v in Visit.query().order(
            -Visit.timestamp).fetch(limit))

升级后：

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

3. 添加任务处理程序（任务运行时调用的代码）

虽然在 fetch_visits() 中可以轻松完成旧版访问的删除，但请注意，此功能与最终用户关系不大。它是一种辅助功能，非常适合在标准应用请求之外异步处理。由于 Datastore 中的信息较少，最终用户将受益于更快的查询速度。创建一个新函数 trim()，该函数通过任务队列 POST 请求调用 /trim，并执行以下操作：

提取“最早的访问”时间戳载荷
发出 Datastore 查询，以查找早于相应时间戳的所有实体。
选择更快的“仅限键”查询，因为不需要实际的用户数据。
记录要删除的实体数量（包括零）。
调用 ndb.delete_multi() 以删除任何实体（如果不存在，则跳过）。
返回一个空字符串（以及一个隐式 HTTP 200 返回代码）。

您可以在下方的 trim() 中查看所有这些内容。将其添加到 main.py 中的 fetch_visits() 之后：

@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

4. 更新网站模板

使用此 Jinja2 条件更新 Web 模板 templates/index.html，以在相应变量存在时显示最早的时间戳：

{% if oldest is defined %}
    <b>Deleting visits older than:</b> {{ oldest }}</p>
{% endif %}

将此代码段添加到显示的访问次数列表之后，但在关闭正文之前，以便您的模板如下所示：

<!doctype html>
<html>
<head>
<title>VisitMe Example</title>
<body>

<h1>VisitMe example</h1>
<h3>Last 10 visits</h3>
<ul>
{% for visit in visits %}
    <li>{{ visit.timestamp.ctime() }} from {{ visit.visitor }}</li>
{% endfor %}
</ul>

{% if oldest is defined %}
    <b>Deleting visits older than:</b> {{ oldest }}</p>
{% endif %}
</body>
</html>

6. 总结/清理

本部分将通过部署应用来结束此 Codelab，并验证应用是否按预期运行以及是否在任何反映的输出中正常运行。应用验证完成后，请执行任何清理操作，并考虑后续步骤。

部署并验证应用

使用 gcloud app deploy 部署应用。输出应与模块 1 应用的输出相同，只是底部会新增一行，显示将删除哪些访问记录：

恭喜您完成此 Codelab。您的代码现在应与模块 7 Repo 文件夹中的内容相匹配。现在，您可以在模块 8 中迁移到 Cloud Tasks 了。

清理

常规

如果您暂时不想继续操作，建议您停用 App Engine 应用，以免产生结算费用。不过，如果您想进一步测试或实验，App Engine 平台有免费配额，因此只要您不超过该使用层级，就不会产生费用。这是计算费用，但相关 App Engine 服务也可能会产生费用，因此请查看其价格页面了解详情。如果此迁移涉及其他 Cloud 服务，则这些服务会单独计费。在任何一种情况下，如果适用，请参阅下文中的“本 Codelab 特有的问题”部分。

为了完全公开透明，我们在此说明，部署到 Google Cloud 无服务器计算平台（例如 App Engine）会产生少量 build 和存储费用。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*ation，例如，如果您的应用托管在美国，则为“us”。

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

此 Codelab 特有的

以下列出的服务是此 Codelab 特有的。如需了解详情，请参阅各个产品的文档：

根据旧版捆绑式服务（例如任务队列）的价格页面，App Engine 任务队列服务不会产生任何额外费用。
App Engine Datastore 服务由 Cloud Datastore（Datastore 模式的 Cloud Firestore）提供，后者也提供免费层级；如需了解详情，请参阅其价格页面。

后续步骤

在此“迁移”中，您向模块 1 示例应用添加了 Task Queue 推送队列使用情况，从而添加了对跟踪访问者的支持，最终得到了模块 7 示例应用。下一个迁移将介绍如何从 App Engine 推送任务升级到 Cloud Tasks（如果您选择这样做）。截至 2021 年秋季，用户在升级到 Python 3 时不再需要迁移到 Cloud Tasks。如需详细了解，请参阅下一部分。

如果您想改用 Cloud Tasks，请继续学习模块 8 Codelab。除此之外，您还需要考虑其他迁移，例如 Cloud Datastore、Cloud Memorystore、Cloud Storage 或 Cloud Pub/Sub（拉取队列）。此外，还可以跨产品迁移到 Cloud Run 和 Cloud Functions。您可以在开源代码库中访问所有无服务器迁移站内容（Codelab、视频、源代码 [如有]）。

7. 迁移到 Python 3

2021 年秋季，App Engine 团队将许多捆绑服务的支持范围扩展到第二代运行时（最初仅在第一代运行时中提供），这意味着在将应用移植到 Python 3 时，您不再需要从 App Engine Task Queue 等捆绑服务迁移到 Cloud Tasks 等独立的 Cloud 或第三方等效服务。换句话说，只要您改造代码以从新一代运行时访问捆绑服务，就可以继续在 Python 3 App Engine 应用中使用任务队列。

您可以在模块 17 Codelab 及其对应的视频中详细了解如何将捆绑服务使用情况迁移到 Python 3。虽然该主题不在模块 7 的范围内，但下面链接的模块 1 和模块 7 应用的 Python 3 版本已移植到 Python 3，并且仍在使用 App Engine NDB 和任务队列。

8. 其他资源

下面列出了其他资源，可供开发者进一步探索本迁移模块或相关迁移模块以及相关产品。这包括提供有关此内容的反馈的途径、指向代码的链接，以及您可能会觉得有用的各种文档。

Codelab 问题/反馈

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

迁移时可参考的资源

下表中提供了指向模块 2（开始）和模块 7（完成）的代码库文件夹的链接。

Codelab	Python 2	Python 3
模块 1	代码	代码（本教程中未介绍）
第 7 模块（本 Codelab）	代码	代码（本教程中未介绍）