App Engine タスクキューの pull タスクから Cloud Pub/Sub に移行する（モジュール 19）

1. 概要

サーバーレス移行ステーションシリーズの Codelab（ご自分のペースで進められる実践型のチュートリアル）と関連動画は、主にレガシーサービスからの移行を 1 つ以上行うことで、Google Cloud サーバーレスのデベロッパーがアプリケーションをモダナイズできるよう支援することを目的としています。これにより、アプリの移植性が向上し、オプションと柔軟性が高まるため、幅広い Cloud プロダクトとの統合やアクセスが可能になり、新しい言語リリースへのアップグレードが容易になります。このシリーズは、当初は初期の Cloud ユーザー、主に App Engine（標準環境）のデベロッパーを対象としていましたが、Cloud Functions や Cloud Run などの他のサーバーレスプラットフォームや、該当する場合は他のプラットフォームも対象とするほど幅広くなっています。

この Codelab の目的は、Python 2 App Engine デベロッパーが App Engine タスクキューの pull タスクから Cloud Pub/Sub に移行する方法を示すことです。また、Datastore アクセス（主にモジュール 2 で説明）のための App Engine NDB から Cloud NDB への暗黙的な移行と、Python 3 へのアップグレードもあります。

モジュール 18 では、アプリでプルタスクを追加する方法を学びました。このモジュールでは、完成したモジュール 18 のアプリを使用して、その使用方法を Cloud Pub/Sub に移行します。push タスクにタスクキューを使用している場合は、代わりに Cloud Tasks に移行します。モジュール 7 ～ 9 を参照してください。

GCP コンソールの

App Engine Task Queue（pull タスク）の使用を Cloud Pub/Sub に置き換える
App Engine NDB の使用を Cloud NDB に置き換えます（モジュール 2 もご覧ください）。
アプリを Python 3 に移植する

必要なもの

有効な GCP 請求先アカウントを持つ Google Cloud Platform プロジェクト
基本的な Python スキル
一般的な Linux コマンドに関する実践的な知識
App Engine アプリの開発とデプロイに関する基礎知識
正常に機能するモジュール 18 App Engine サンプルアプリ

アンケート

このチュートリアルの利用方法をお選びください。

全体を通して読むだけ

内容を読んで演習をやり遂げる

Python のご利用経験はどの程度ありますか？

初心者

中級者

上級者

Google Cloud サービスの使用経験はどの程度ありますか？

初心者

中級者

上級者

2. 背景情報

App Engine Task Queue は、push タスクと pull タスクの両方をサポートしています。アプリケーションの移植性を高めるため、Google Cloud では、Task Queue などの以前のバンドルサービスから、他の Cloud スタンドアロンサービスまたはサードパーティの同等のサービスに移行することをおすすめします。

Task Queue の push タスクのユーザーは、Cloud Tasks に移行する必要があります。
Task Queue の pull タスクのユーザーは、Cloud Pub/Sub に移行する必要があります。

移行モジュール 7 ～ 9 ではプッシュタスクの移行について説明し、モジュール 18 ～ 19 ではプルタスクの移行について説明します。Cloud Tasks はタスクキューの push タスクとより密接に一致しますが、Pub/Sub はタスクキューの pull タスクとそれほど類似していません。

Pub/Sub には、Task Queue が提供する pull 機能よりも多くの機能があります。たとえば、Pub/Sub には push 機能もありますが、Cloud Tasks は Task Queue の push タスクに似ているため、Pub/Sub の push は移行モジュールでは 対象外です。このモジュール 19 の Codelab では、キューイングメカニズムをタスクキューの pull キューから Pub/Sub に切り替え、Datastore アクセス用に App Engine NDB から Cloud NDB に移行し、モジュール 2 の移行を繰り返す方法を説明します。

モジュール 18 のコードは Python 2 サンプルアプリとして「宣伝」されていますが、ソース自体は Python 2 と 3 の両方に対応しており、モジュール 19 で Cloud Pub/Sub（と Cloud NDB）に移行した後もその状態が維持されます。

このチュートリアルでは、次の手順について説明します。

セットアップ / 事前作業
構成の更新
アプリケーションコードを変更する

3. セットアップ/事前作業

このセクションでは、次の方法を説明します。

Cloud プロジェクトを設定する
ベースラインサンプルアプリを入手する
ベースラインアプリを（再）デプロイして検証する
新しい Google Cloud サービス/API を有効にする

これらの手順により、動作するコードから開始し、Cloud サービスへの移行の準備が整います。

1. プロジェクトのセットアップ

モジュール 18 Codelab を完了している場合は、同じプロジェクト（とコード）を再利用します。または、新しいプロジェクトを作成するか、別の既存のプロジェクトを再利用することもできます。プロジェクトにアクティブな請求先アカウントがあり、App Engine アプリが有効になっていることを確認します。この Codelab では、PROJECT_ID 変数が表示されるたびにプロジェクト ID を使用するため、プロジェクト ID を確認しておきます。

2. ベースラインサンプルアプリを入手する

前提条件の 1 つは、機能するモジュール 18 の App Engine アプリです。そのため、Codelab（推奨、上のリンク）を完了するか、リポジトリからモジュール 18 のコードをコピーします。ご自分のコードと Google で用意したコードのいずれを使用される場合も、ここから開始します（「START」）。この Codelab では、移行の手順を説明します。完了すると、モジュール 19 のリポジトリフォルダ（「FINISH」）にあるコードに似たコードが生成されます。

開始: モジュール 18 フォルダ（Python 2）
終了: モジュール 19 フォルダ（Python 2 と 3）
リポジトリ全体（クローン作成または ZIP ファイルのダウンロード）

モジュール 18 のどのアプリを使用しても、フォルダは次のようになります（lib フォルダも含まれる場合があります）。

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

3. ベースラインアプリを（再）デプロイして検証する

モジュール 18 のアプリをデプロイする手順は次のとおりです。

lib フォルダが存在する場合は削除し、pip install -t lib -r requirements.txt を実行して lib を再作成します。開発マシンに Python 2 と 3 の両方がインストールされている場合は、代わりに pip2 を使用する必要があるかもしれません。
gcloud コマンドラインツールをインストールして初期化し、その使用方法を確認してください。
（省略可）発行する gcloud コマンドごとに PROJECT_ID を入力したくない場合は、gcloud config set project PROJECT_ID を使用して Cloud プロジェクトを設定します。
gcloud app deploy を使用してサンプルアプリをデプロイする
アプリが問題なく想定どおりに動作することを確認します。モジュール 18 の Codelab を完了している場合、アプリには上位の訪問者と最新の訪問が表示されます（下図を参照）。そうでない場合、表示する訪問者数が存在しない可能性があります。

モジュール 18 のサンプルアプリを移行する前に、変更されたアプリで使用する Cloud サービスを有効にする必要があります。

4. 新しい Google Cloud サービス/API を有効にする

以前のアプリでは、追加の設定を必要としない App Engine バンドルサービスを使用していましたが、スタンドアロンの Cloud サービスでは追加の設定が必要です。更新されたアプリでは、Cloud Pub/Sub と Cloud Datastore（Cloud NDB クライアントライブラリ経由）の両方が使用されます。App Engine と両方の Cloud API には「常時無料」枠の割り当てがあります。この上限を超えない限り、このチュートリアルを完了しても料金は発生しません。Cloud API は、必要に応じて Cloud コンソールまたはコマンドラインから有効にできます。

Cloud Console から

Cloud Console で（正しいプロジェクトの）API Manager の [ライブラリ] ページに移動し、ページの中央にある検索バーを使用して Cloud Datastore API と Cloud Pub/Sub API を検索します。

各 API の [有効にする] ボタンを個別にクリックします。お支払い情報の入力を求められることがあります。たとえば、Cloud Pub/Sub API ライブラリのページは次のようになります。

コマンドラインから

コンソールから API を有効にすると、視覚的にわかりやすいですが、コマンドラインを好むユーザーもいます。gcloud services enable pubsub.googleapis.com datastore.googleapis.com コマンドを発行して、両方の API を同時に有効にします。

$ gcloud services enable pubsub.googleapis.com datastore.googleapis.com
Operation "operations/acat.p2-aaa-bbb-ccc-ddd-eee-ffffff" finished successfully.

お支払い情報の入力を求められることがあります。他の Cloud APIs を有効にして、その URI を確認する場合は、各 API のライブラリページの下部で確認できます。たとえば、上の Pub/Sub ページの下部にある [サービス名] として pubsub.googleapis.com を確認します。

手順が完了すると、プロジェクトから API にアクセスできるようになります。次に、これらの API を使用するようにアプリケーションを更新します。

4. Pub/Sub リソースを作成する

モジュール 18 のタスクキューワークフローのシーケンス順序をまとめると、次のようになります。

モジュール 18 では、queue.yaml ファイルを使用して pullq という名前の pull キューを作成しました。
アプリは pull キューにタスクを追加して、訪問者を追跡します。
タスクは最終的にワーカーによって処理され、一定時間（1 時間）リースされます。
タスクが実行され、最近の訪問者数が集計されます。
タスクは完了するとキューから削除されます。

Pub/Sub を使用して同様のワークフローを複製します。次のセクションでは、基本的な Pub/Sub 用語と、必要な Pub/Sub リソースを作成する 3 つの方法について説明します。

App Engine Task Queue（pull）と Cloud Pub/Sub の用語

Pub/Sub に切り替えるには、語彙を少し調整する必要があります。以下に、両方のプロダクトの関連用語とともに、主なカテゴリを記載します。同様の比較が記載されている移行ガイドもご確認ください。

キューイングデータ構造: Task Queue では、データは pull キューに送られます。Pub/Sub では、データはトピックに送られます。
キューに登録されたデータの単位: Task Queue の pull タスクは、Pub/Sub ではメッセージと呼ばれます。
データプロセッサ: Task Queue では、ワーカーが pull タスクにアクセスします。Pub/Sub では、メッセージを受信するためにサブスクリプション/サブスクライバーが必要です。
データ抽出: pull タスクのリースは、トピックから（サブスクリプションを介して）メッセージを pull するのと同じです。
クリーンアップ/完了: 完了時に pull キューから Task Queue タスクを削除することは、Pub/Sub メッセージを確認応答することに似ています。

キューイングプロダクトは変更されますが、ワークフローは比較的似ています。

アプリは、プルキューではなく pullq という名前のトピックを使用します。
アプリは、タスクを pull キューに追加するのではなく、トピック（pullq）にメッセージを送信します。
ワーカーが pull キューからタスクをリースするのではなく、worker という名前のサブスクライバーが pullq トピックからメッセージを pull します。
アプリはメッセージペイロードを処理し、Datastore の訪問者数を増やします。
アプリは、pull キューからタスクを削除するのではなく、処理済みのメッセージを確認応答します。

Task Queue の設定では、pull キューを作成します。Pub/Sub では、設定を行うにはトピックとサブスクリプションの両方を作成する必要があります。モジュール 18 では、アプリの実行外で queue.yaml を処理しました。今度は Pub/Sub で同じことを行う必要があります。

トピックとサブスクリプションを作成するには、次の 3 つの方法があります。

Cloud コンソールから
コマンドラインから、または
コードから（短い Python スクリプト）

次のいずれかのオプションを選択し、対応する手順に沿って Pub/Sub リソースを作成します。

Cloud コンソールから

Cloud Console からトピックを作成する手順は次のとおりです。

Cloud コンソールの Pub/Sub トピックページに移動します。
上部の [トピックを作成] をクリックします。新しいダイアログウィンドウが開きます（下の画像を参照）。
[トピック ID] フィールドに「pullq」と入力します。
オンになっているオプションをすべてオフにして、[Google が管理する暗号鍵] を選択します。
[トピックを作成] ボタンをクリックします。

トピック作成ダイアログは次のようになります。

トピックが作成されたので、そのトピックのサブスクリプションを作成する必要があります。

Cloud コンソールの Pub/Sub サブスクリプションページに移動します。
上部の [サブスクリプションを作成] をクリックします（下の画像を参照）。
[サブスクリプション ID] フィールドに「worker」と入力します。
[Cloud Pub/Sub トピックを選択してください] プルダウンから pullq を選択し、その「完全修飾パス名」（例: projects/PROJECT_ID/topics/pullq）をメモします。
[配信タイプ] で、[pull] を選択します。
他のオプションはすべてそのままにして、[作成] ボタンをクリックします。

定期購入の作成画面は次のようになります。

[トピック] ページからサブスクリプションを作成することもできます。この「ショートカット」は、トピックとサブスクリプションの関連付けに役立ちます。サブスクリプションの作成について詳しくは、ドキュメントをご覧ください。

コマンドラインから

Pub/Sub ユーザーは、それぞれ gcloud pubsub topics create TOPIC_ID コマンドと gcloud pubsub subscriptions create SUBSCRIPTION_ID --topic=TOPIC_ID コマンドを使用して、トピックとサブスクリプションを作成できます。TOPIC_ID が pullq、SUBSCRIPTION_ID が worker の状態でこれらのコマンドを実行すると、プロジェクト PROJECT_ID の出力は次のようになります。

$ gcloud pubsub topics create pullq
Created topic [projects/PROJECT_ID/topics/pullq].

$ gcloud pubsub subscriptions create worker --topic=pullq
Created subscription [projects/PROJECT_ID/subscriptions/worker].

クイックスタートドキュメントのこちらのページもご覧ください。コマンドラインを使用すると、トピックとサブスクリプションが定期的に作成されるワークフローを簡素化できます。このようなコマンドは、この目的でシェルスクリプトで使用できます。

コードから（短い Python スクリプト）

トピックとサブスクリプションの作成を自動化する別の方法は、ソースコードで Pub/Sub API を使用することです。以下は、Module 19 のリポジトリフォルダにある maker.py スクリプトのコードです。

from __future__ import print_function
import google.auth
from google.api_core import exceptions
from google.cloud import pubsub

_, PROJECT_ID = google.auth.default()
TOPIC = 'pullq'
SBSCR = 'worker'
ppc_client = pubsub.PublisherClient()
psc_client = pubsub.SubscriberClient()
TOP_PATH = ppc_client.topic_path(PROJECT_ID, TOPIC)
SUB_PATH = psc_client.subscription_path(PROJECT_ID, SBSCR)

def make_top():
    try:
        top = ppc_client.create_topic(name=TOP_PATH)
        print('Created topic %r (%s)' % (TOPIC, top.name))
    except exceptions.AlreadyExists:
        print('Topic %r already exists at %r' % (TOPIC, TOP_PATH))

def make_sub():
    try:
        sub = psc_client.create_subscription(name=SUB_PATH, topic=TOP_PATH)
        print('Subscription created %r (%s)' % (SBSCR, sub.name))
    except exceptions.AlreadyExists:
        print('Subscription %r already exists at %r' % (SBSCR, SUB_PATH))
    try:
        psc_client.close()
    except AttributeError:  # special Py2 handler for grpcio<1.12.0
        pass

make_top()
make_sub()

このスクリプトを実行すると、エラーがなければ、想定どおりの出力が得られます。

$ python3 maker.py
Created topic 'pullq' (projects/PROJECT_ID/topics/pullq)
Subscription created 'worker' (projects/PROJECT_ID/subscriptions/worker)

API を呼び出して既存のリソースを作成すると、クライアントライブラリによって google.api_core.exceptions.AlreadyExists 例外がスローされます。この例外はスクリプトによって適切に処理されます。

$ python3 maker.py
Topic 'pullq' already exists at 'projects/PROJECT_ID/topics/pullq'
Subscription 'worker' already exists at 'projects/PROJECT_ID/subscriptions/worker'

Pub/Sub を初めて使用する場合は、Pub/Sub アーキテクチャのホワイトペーパーで詳細をご確認ください。

5. 構成を更新する

構成の更新には、さまざまな構成ファイルの変更と、App Engine pull キューと同等のものを Cloud Pub/Sub エコシステム内に作成することが含まれます。

queue.yaml を削除する

タスクキューを完全に移行するため、Pub/Sub でこのファイルが使用されないため、queue.yaml を削除します。pull キューを作成する代わりに、Pub/Sub トピック（とサブスクリプション）を作成します。

requirements.txt

google-cloud-ndb と google-cloud-pubsub の両方を requirements.txt に追加して、モジュール 18 から flask に参加します。更新されたモジュール 19 の requirements.txt は次のようになります。

flask
google-cloud-ndb
google-cloud-pubsub

この requirements.txt ファイルにはバージョン番号が含まれていないため、最新バージョンが選択されます。互換性の問題が発生した場合は、バージョン番号を使用してアプリの動作バージョンをロックインするという標準的な方法に従います。

app.yaml

app.yaml の変更は、Python 2 を使用し続けるか、Python 3 にアップグレードするかによって異なります。

Python 2

上記の requirements.txt の更新では、Google Cloud クライアントライブラリの使用が追加されています。これらには、App Engine からの追加サポート（組み込みライブラリの setuptools と grpcio）が必要です。組み込みライブラリを使用するには、app.yaml に libraries セクションとライブラリのバージョン番号が必要です。App Engine サーバーで利用可能な最新バージョンを使用する場合は、「latest」と指定します。モジュール 18 の app.yaml には、まだこれらのセクションのいずれもありません。

変更前:

runtime: python27
threadsafe: yes
api_version: 1

handlers:
- url: /.*
  script: main.app

app.yaml に libraries セクションを追加し、setuptools と grpcio の両方のエントリを追加して、最新バージョンを選択します。また、Python 3 のプレースホルダ runtime エントリを追加し、現在の 3.x リリース（このドキュメントの作成時点では 3.10 など）とともにコメントアウトします。これらの変更を行うと、app.yaml は次のようになります。

変更後:

#runtime: python310
runtime: python27
threadsafe: yes
api_version: 1

handlers:
- url: /.*
  script: main.app

libraries:
- name: setuptools
  version: latest
- name: grpcio
  version: latest

Python 3

Python 3 ユーザーと app.yaml の場合は、削除がすべてです。このセクションでは、handlers セクション、threadsafe ディレクティブ、api_version ディレクティブを削除し、libraries セクションは作成しません。

第 2 世代ランタイムには組み込みのサードパーティライブラリが用意されていないため、app.yaml に libraries セクションは不要です。さらに、組み込みでないサードパーティパッケージのコピー（ベンダリングまたはセルフバンドルとも呼ばれます）は不要になりました。requirements.txt には、アプリが使用するサードパーティ製ライブラリのみをリストする必要があります。

app.yaml の handlers セクションは、アプリケーション（スクリプト）と静的ファイルハンドラを指定するためのものです。Python 3 ランタイムでは、ウェブフレームワークが独自のルーティングを実行する必要があるため、すべてのスクリプトハンドラを auto に変更する必要があります。アプリ（モジュール 18 のアプリなど）が静的ファイルを提供しない場合、すべてのルートが auto になり、無関係になります。その結果、handlers セクションも不要になるため、削除します。

最後に、Python 3 では threadsafe ディレクティブも api_version ディレクティブも使用されないため、これらも削除します。要するに、app.yaml のすべてのセクションを削除して、runtime ディレクティブのみを残し、最新バージョンの Python 3（3.10 など）を指定する必要があります。これらの更新前後の app.yaml の様子は次のとおりです。

変更前:

runtime: python27
threadsafe: yes
api_version: 1

handlers:
- url: /.*
  script: main.app

変更後:

runtime: python310

Python 3 の app.yaml からすべてを削除する準備ができていない場合は、Module 19 のリポジトリフォルダに app3.yaml の代替ファイルを用意しています。デプロイにこれを使用する場合は、コマンドの末尾にファイル名 gcloud app deploy app3.yaml を追加してください（追加しない場合、デフォルトで Python 2 の app.yaml ファイルが使用され、アプリがデプロイされます）。

appengine_config.py

Python 3 にアップグレードする場合は、appengine_config.py は不要なので削除します。サードパーティライブラリのサポートでは、requirements.txt で指定するだけで済むため、必要ありません。Python 2 をご利用の場合は、以下をお読みください。

モジュール 18 の appengine_config.py には、サードパーティライブラリをサポートする適切なコードがあります。たとえば、Flask や、requirements.txt に追加されたばかりの Cloud クライアントライブラリなどです。

変更前:

from google.appengine.ext import vendor

# Set PATH to your libraries folder.
PATH = 'lib'
# Add libraries installed in the PATH folder.
vendor.add(PATH)

ただし、このコードだけでは、追加された組み込みライブラリ（setuptools、grpcio）をサポートするには不十分です。数行のコードを追加する必要があるため、appengine_config.py を次のように更新します。

変更後:

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)

Cloud クライアントライブラリをサポートするために必要な変更の詳細については、バンドルサービスの移行に関するドキュメントをご覧ください。

その他の構成の更新

lib フォルダがある場合は、削除します。Python 2 を使用している場合は、次のコマンドを実行して lib フォルダを補充します。

pip install -t lib -r requirements.txt  # or pip2

開発システムに Python 2 と 3 の両方がインストールされている場合は、pip ではなく pip2 を使用する必要がある場合があります。

6. アプリケーションコードを変更する

このセクションでは、App Engine タスクキューの pull キューの使用を Cloud Pub/Sub に置き換える、メインアプリケーションファイル main.py の更新について説明します。ウェブテンプレート templates/index.html に変更はありません。どちらのアプリも同じように動作し、同じデータを表示します。

インポートと初期化を更新する

インポートと初期化にはいくつかの更新があります。

インポートの場合、App Engine NDB と Task Queue を Cloud NDB と Pub/Sub に置き換えます。
pullq の名前を QUEUE から TOPIC に変更します。
pull タスクでは、ワーカーは 1 時間リースしていましたが、Pub/Sub ではタイムアウトはメッセージごとに測定されるため、HOUR 定数を削除します。
Cloud API では API クライアントを使用する必要があるため、Cloud NDB と Cloud Pub/Sub の API クライアントを初期化します。後者では、トピックとサブスクリプションの両方のクライアントが提供されます。
Pub/Sub には Cloud プロジェクト ID が必要なので、google.auth.default() からインポートして取得します。
Pub/Sub では、トピックとサブスクリプションに「完全修飾パス名」が必要なため、*_path() 簡易関数を使用して作成します。

以下は、モジュール 18 のインポートと初期化です。その後に、上記の変更を実装した後のセクションの様子を示します。新しいコードのほとんどは、さまざまな Pub/Sub リソースです。

変更前:

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

HOUR = 3600
LIMIT = 10
TASKS = 1000
QNAME = 'pullq'
QUEUE = taskqueue.Queue(QNAME)
app = Flask(__name__)

変更後:

from flask import Flask, render_template, request
import google.auth
from google.cloud import ndb, pubsub

LIMIT = 10
TASKS = 1000
TOPIC = 'pullq'
SBSCR = 'worker'

app = Flask(__name__)
ds_client  = ndb.Client()
ppc_client = pubsub.PublisherClient()
psc_client = pubsub.SubscriberClient()
_, PROJECT_ID = google.auth.default()
TOP_PATH = ppc_client.topic_path(PROJECT_ID, TOPIC)
SUB_PATH = psc_client.subscription_path(PROJECT_ID, SBSCR)

データモデルの更新にアクセスする

Visit データモデルは変更されません。Datastore にアクセスするには、Cloud NDB API クライアントコンテキストマネージャー ds_client.context() を明示的に使用する必要があります。コードでは、Python の with ブロック内の store_visit() と fetch_visits() の両方で Datastore 呼び出しをラップします。この更新は、モジュール 2 で取り上げた内容と同じです。

Pub/Sub に関連する最も重要な変更は、Task Queue の pull タスクのキュー登録を、pullq トピックへの Pub/Sub メッセージのパブリッシュに置き換えることです。以下は、これらの更新を行う前後のコードです。

変更前:

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 in Datastore and queue request to bump visitor count'
    Visit(visitor='{}: {}'.format(remote_addr, user_agent)).put()
    QUEUE.add(taskqueue.Task(payload=remote_addr, method='PULL'))

def fetch_visits(limit):
    'get most recent visits'
    return Visit.query().order(-Visit.timestamp).fetch(limit)

変更後:

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 in Datastore and queue request to bump visitor count'
    with ds_client.context():
        Visit(visitor='{}: {}'.format(remote_addr, user_agent)).put()
    ppc_client.publish(TOP_PATH, remote_addr.encode('utf-8'))

def fetch_visits(limit):
    'get most recent visits'
    with ds_client.context():
        return Visit.query().order(-Visit.timestamp).fetch(limit)

VisitorCount データモデルの更新

VisitorCount データモデルは変更されず、次の図に示すように、Datastore クエリを with ブロックでラップする以外は fetch_counts() を実行します。

変更前:

class VisitorCount(ndb.Model):
    visitor = ndb.StringProperty(repeated=False, required=True)
    counter = ndb.IntegerProperty()

def fetch_counts(limit):
    'get top visitors'
    return VisitorCount.query().order(-VisitorCount.counter).fetch(limit)

変更後:

class VisitorCount(ndb.Model):
    visitor = ndb.StringProperty(repeated=False, required=True)
    counter = ndb.IntegerProperty()

def fetch_counts(limit):
    'get top visitors'
    with ds_client.context():
        return VisitorCount.query().order(-VisitorCount.counter).fetch(limit)

ワーカーコードを更新する

ワーカーコードは、NDB を Cloud NDB に、Task Queue を Pub/Sub に置き換えるまで更新されますが、ワークフローは同じままです。

Cloud NDB コンテキストマネージャー with ブロックで Datastore 呼び出しをラップします。
タスクキューのクリーンアップでは、プルキューからすべてのタスクを削除します。Pub/Sub では、確認応答 ID は acks に収集され、最後に削除または確認応答されます。
タスクキューの pull タスクは、Pub/Sub メッセージが pull されるのと同様の方法でリースされます。pull タスクの削除はタスクオブジェクト自体で行われますが、Pub/Sub メッセージは確認応答 ID を介して削除されます。
Pub/Sub メッセージペイロードにはバイト（Python 文字列ではない）が必要なため、トピックへのパブリッシュ時とトピックからのメッセージの取得時に、それぞれ UTF-8 エンコードとデコードが行われます。

log_visitors() を、上記で説明した変更を実装した次の更新済みコードに置き換えます。

変更前:

@app.route('/log')
def log_visitors():
    'worker processes recent visitor counts and updates them in Datastore'
    # tally recent visitor counts from queue then delete those tasks
    tallies = {}
    tasks = QUEUE.lease_tasks(HOUR, TASKS)
    for task in tasks:
        visitor = task.payload
        tallies[visitor] = tallies.get(visitor, 0) + 1
    if tasks:
        QUEUE.delete_tasks(tasks)

    # increment those counts in Datastore and return
    for visitor in tallies:
        counter = VisitorCount.query(VisitorCount.visitor == visitor).get()
        if not counter:
            counter = VisitorCount(visitor=visitor, counter=0)
            counter.put()
        counter.counter += tallies[visitor]
        counter.put()
    return 'DONE (with %d task[s] logging %d visitor[s])\r\n' % (
            len(tasks), len(tallies))

変更後:

@app.route('/log')
def log_visitors():
    'worker processes recent visitor counts and updates them in Datastore'
    # tally recent visitor counts from queue then delete those tasks
    tallies = {}
    acks = set()
    rsp = psc_client.pull(subscription=SUB_PATH, max_messages=TASKS)
    msgs = rsp.received_messages
    for rcvd_msg in msgs:
        acks.add(rcvd_msg.ack_id)
        visitor = rcvd_msg.message.data.decode('utf-8')
        tallies[visitor] = tallies.get(visitor, 0) + 1
    if acks:
        psc_client.acknowledge(subscription=SUB_PATH, ack_ids=acks)
    try:
        psc_client.close()
    except AttributeError:  # special handler for grpcio<1.12.0
        pass

    # increment those counts in Datastore and return
    if tallies:
        with ds_client.context():
            for visitor in tallies:
                counter = VisitorCount.query(VisitorCount.visitor == visitor).get()
                if not counter:
                    counter = VisitorCount(visitor=visitor, counter=0)
                    counter.put()
                counter.counter += tallies[visitor]
                counter.put()
    return 'DONE (with %d task[s] logging %d visitor[s])\r\n' % (
            len(msgs), len(tallies))

メインアプリケーションハンドラ root() の変更はありません。HTML テンプレートファイル templates/index.html も変更する必要がないため、必要な更新はすべて完了します。Cloud Pub/Sub を使用した新しいモジュール 19 アプリケーションに到達しました。おめでとうございます。

7. 概要/クリーンアップ

アプリをデプロイして、意図したとおりに動作することと、反映された出力で動作することを確認します。また、ワーカーを実行して訪問者数を処理します。アプリの検証が完了したら、クリーンアップの手順を実行し、次の手順を検討します。

アプリケーションをデプロイして検証する

pullq トピックと worker サブスクリプションがすでに作成されていることを確認します。完了していて、サンプルアプリの準備ができている場合は、gcloud app deploy を使用してアプリをデプロイします。出力は、基盤となるキューイングメカニズム全体が正常に置き換えられたことを除いて、モジュール 18 のアプリと同じになります。

アプリのウェブフロントエンドが、アプリケーションのこの部分が動作することを確認します。アプリのこの部分では、上位の訪問者と最近の訪問を正常にクエリして表示しますが、アプリは訪問を登録し、この訪問者を合計数に追加する pull タスクを作成します。そのタスクは現在、処理待ちのキューにあります。

これは、App Engine バックエンドサービス、cron ジョブ、/log へのブラウジング、コマンドライン HTTP リクエストの発行によって実行できます。curl を使用してワーカーコードを呼び出す実行例と出力例を次に示します（PROJECT_ID はご自身のものに置き換えてください）。

$ curl https://PROJECT_ID.appspot.com/log
DONE (with 1 task[s] logging 1 visitor[s])

更新されたカウントは、次回ウェブサイトにアクセスしたときに反映されます。これを探してた！

クリーンアップ

全般

現時点で完了した場合は、課金が発生しないように 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 Pub/Sub のさまざまなコンポーネントには無料枠があります。費用の影響を把握するには、全体的な使用量を把握し、詳細については料金ページをご覧ください。
App Engine Datastore サービスは Cloud Datastore（Datastore モードの Cloud Firestore）によって提供されます。これにも無料枠があります。詳細については、料金ページをご覧ください。

次のステップ

このチュートリアル以外にも、以前のバンドルサービスからの移行に焦点を当てた移行モジュールがあります。

モジュール 2: App Engine ndb から Cloud NDB に移行する
モジュール 7 ～ 9: App Engine Task Queue（push タスク）から Cloud Tasks に移行する
モジュール 12 ～ 13: App Engine Memcache から Cloud Memorystore への移行
モジュール 15 ～ 16: App Engine Blobstore から Cloud Storage への移行

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 を使用せずにコンテナ化します。

別のサーバーレスプラットフォームへの切り替えは任意です。変更を行う前に、アプリとユースケースに最適なオプションを検討することをおすすめします。

次にどの移行モジュールを検討する場合でも、すべてのサーバーレス移行ステーションのコンテンツ（Codelab、動画、ソースコード [利用可能な場合]）は、オープンソースリポジトリからアクセスできます。リポジトリの README には、考慮すべき移行と、移行モジュールの関連する「順序」に関するガイダンスも記載されています。

8. 参考情報

以下に、デベロッパーがこの移行モジュールまたは関連する移行モジュールや関連プロダクトをさらに詳しく調べるための追加リソースを示します。このコンテンツに関するフィードバックを送信できる場所、コードへのリンク、役立つ可能性のあるさまざまなドキュメントなどが含まれます。

Codelab に関する問題/フィードバック

この Codelab に問題が見つかった場合は、提出する前にまず問題を検索してください。新しい問題の検索と登録を行うためのリンク:

移行に関するリソース

モジュール 18（開始）とモジュール 19（終了）のリポジトリフォルダへのリンクを以下の表に示します。

Codelab	Python 2	Python 3
モジュール 18	コード	該当なし
モジュール 19（この Codelab）	コード	（上記で説明したように app.yaml を更新した場合を除き、Python 2 と同じですが、app3.yaml を使用します）

オンラインリファレンス

このチュートリアルに関連するリソースは次のとおりです。

App Engine のタスクキュー

Cloud Pub/Sub

Cloud Pub/Sub のプロダクトページ
Pub/Sub クライアントライブラリの使用
Pub/Sub Python クライアントライブラリのサンプル
Pub/Sub Python クライアントライブラリのドキュメント
Pub/Sub トピックの作成と管理
Pub/Sub トピックの命名ガイドライン
Pub/Sub サブスクリプションの作成と管理
App Engine（フレキシブル）サンプルアプリ（スタンダードにもデプロイ可能、Python 3）
上記のサンプルアプリのリポジトリ
Pub/Sub pull サブスクリプション
Pub/Sub push サブスクリプション
App Engine Pub/Sub プッシュサンプルアプリ（Python 3）
App Engine Pub/Sub プッシュサンプルアプリリポジトリ
Pub/Sub の料金情報
Cloud Tasks と Cloud Pub/Sub の選択（プッシュとプル）

App Engine NDB と Cloud NDB（Datastore）

App Engine プラットフォーム

その他のクラウド情報

動画

ライセンス

この作業はクリエイティブ・コモンズの表示 2.0 汎用ライセンスにより使用許諾されています。

App Engine タスクキューの pull タスクから Cloud Pub/Sub に移行する（モジュール 19）

1. 概要

GCP コンソールの

必要なもの

アンケート

このチュートリアルの利用方法をお選びください。

Python のご利用経験はどの程度ありますか？

Google Cloud サービスの使用経験はどの程度ありますか？

2. 背景情報

3. セットアップ/事前作業

1. プロジェクトのセットアップ

2. ベースライン サンプルアプリを入手する

3. ベースライン アプリを（再）デプロイして検証する

4. 新しい Google Cloud サービス/API を有効にする

Cloud Console から

コマンドラインから

4. Pub/Sub リソースを作成する

App Engine Task Queue（pull）と Cloud Pub/Sub の用語

Cloud コンソールから

コマンドラインから

コードから（短い Python スクリプト）

5. 構成を更新する

queue.yaml を削除する

requirements.txt

app.yaml

Python 2

Python 3

appengine_config.py

その他の構成の更新

6. アプリケーション コードを変更する

インポートと初期化を更新する

データモデルの更新にアクセスする

VisitorCount データモデルの更新

ワーカーコードを更新する

7. 概要/クリーンアップ

アプリケーションをデプロイして検証する

クリーンアップ

全般

この Codelab に固有

次のステップ

8. 参考情報

Codelab に関する問題/フィードバック

移行に関するリソース

オンライン リファレンス

App Engine のタスクキュー

Cloud Pub/Sub

App Engine NDB と Cloud NDB（Datastore）

App Engine プラットフォーム

その他のクラウド情報

動画

ライセンス

2. ベースラインサンプルアプリを入手する

3. ベースラインアプリを（再）デプロイして検証する

6. アプリケーションコードを変更する

オンラインリファレンス