Flask アプリで App Engine タスクキュー（pull タスク）を使用する方法（モジュール 18）

1. 概要

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

この Codelab では、App Engine タスクキューの pull タスクを モジュール 1 の Codelab の サンプルアプリに含めて使用する方法を学びます。この Module 18 のチュートリアルでは、pull タスクの使用を追加し、Module 19 でその使用を Cloud Pub/Sub に移行します。push タスクにタスクキューを使用している場合は、代わりに Cloud Tasks に移行します。モジュール 7 ～ 9 を参照してください。

GCP コンソールの

  • App Engine Task Queue API/バンドル サービスを使用する
  • 基本的な Python 2 Flask App Engine NDB アプリに pull キューの使用を追加する

必要なもの

アンケート

このチュートリアルをどのように使用されますか？

通読のみ 通読して演習を行う

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

初心者 中級者 上級者

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

初心者 中級者 上級者

2. 背景情報

App Engine Task Queue pull タスクから移行するには、モジュール 1 の Codelab で作成した既存の Flask アプリと App Engine NDB アプリに、その使用方法を追加します。サンプルアプリには、エンドユーザーの最新のアクセスが表示されます。それもよいのですが、訪問者をトラッキングして、誰が最も多く訪問しているかを確認するのも興味深いでしょう。

これらの訪問者数にはプッシュタスクを使用できますが、訪問を登録してユーザーにすぐにレスポンスを返すサンプルアプリと、通常のリクエスト / レスポンス ワークフロー外で訪問者数を集計する指定された「ワーカー」との間で責任を分担します。

この設計を実装するため、メイン アプリケーションにプルキューの使用を追加し、ワーカー機能をサポートします。ワーカーは、別のプロセス（バックエンド インスタンスや常に稼働している VM で実行されるコードなど）、cron ジョブ、または curl または wget を使用する基本的なコマンドライン HTTP リクエストとして実行できます。この統合が完了したら、次の Codelab（モジュール 19）でアプリを Cloud Pub/Sub に移行できます。

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

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

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

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

  1. Cloud プロジェクトを設定する
  2. ベースライン サンプルアプリを入手する
  3. ベースライン アプリを（再）デプロイして検証する

これらの手順により、動作するコードから開始できます。

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

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

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

この Codelab の前提条件の 1 つは、機能するモジュール 1 の App Engine アプリを用意することです。モジュール 1 Codelab（推奨）を完了するか、リポジトリからモジュール 1 アプリをコピーします。ご自分のコードと Google で用意したコードのいずれを使用される場合も、モジュール 1 のコードから開始します。この Codelab では、各ステップを順を追って説明します。最後に、モジュール 18 のリポジトリ フォルダ「FINISH」にあるコードに似たコードが完成します。

モジュール 1 のどのアプリを使用しても、フォルダは次の出力のようになります。lib フォルダも含まれている可能性があります。

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

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

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

  1. lib フォルダが存在する場合は削除し、pip install -t lib -r requirements.txt を実行して lib を再作成します。Python 2 と 3 の両方がインストールされている場合は、代わりに pip2 コマンドを使用する必要があります。
  2. gcloud コマンドライン ツールをインストールして初期化し、その使用方法を確認してください。
  3. 発行する gcloud コマンドごとに PROJECT_ID を入力したくない場合は、gcloud config set project PROJECT_ID を使用して Cloud プロジェクトを設定します。
  4. gcloud app deploy を使用してサンプルアプリをデプロイする
  5. モジュール 1 アプリが想定どおりに実行され、最新のアクセスが表示されることを確認します（下図を参照）。

a7a9d2b80d706a2b.png

4. 構成を更新する

標準の App Engine 構成ファイル（app.yaml、requirements.txt、appengine_config.py）を変更する必要はありません。代わりに、次の内容で新しい構成ファイル queue.yaml を追加し、同じ最上位ディレクトリに配置します。

queue:
- name: pullq
  mode: pull

queue.yaml ファイルには、アプリに存在するすべてのタスクキュー（App Engine によって自動的に作成される default [push] キューを除く）が指定されています。この場合、pullq という名前の pull キューが 1 つだけあります。App Engine では、mode ディレクティブを pull として指定する必要があります。指定しない場合は、デフォルトで push キューが作成されます。pull キューの作成について詳しくは、ドキュメントをご覧ください。その他のオプションについては、queue.yaml リファレンス ページもご覧ください。

このファイルはアプリとは別にデプロイします。引き続き gcloud app deploy を使用しますが、コマンドラインで queue.yaml も指定します。

$ gcloud app deploy queue.yaml
Configurations to update:

descriptor:      [/tmp/mod18-gaepull/queue.yaml]
type:            [task queues]
target project:  [my-project]

WARNING: Caution: You are updating queue configuration. This will override any changes performed using 'gcloud tasks'. More details at
https://cloud.google.com/tasks/docs/queue-yaml

Do you want to continue (Y/n)?

Updating config [queue]...⠹WARNING: We are using the App Engine app location (us-central1) as the default location. Please use the "--location" flag if you want to use a different location.
Updating config [queue]...done.

Task queues have been updated.

Visit the Cloud Platform Console Task Queues page to view your queues and cron jobs.
$

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

このセクションでは、次のファイルの更新について説明します。

  • main.py - pull キューの使用をメイン アプリケーションに追加
  • templates/index.html - 新しいデータを表示するようにウェブ テンプレートを更新します。

インポートと定数

まず、プルキューをサポートするために、新しいインポートを 1 つといくつかの定数を追加します。

  • Task Queue ライブラリ google.appengine.api.taskqueue のインポートを追加します。
  • プルキュー（QUEUE）から 1 時間（HOUR）にプルタスク（TASKS）を最大数リースできるように、3 つの定数を追加します。
  • 直近のアクセスと上位のアクセスユーザーを表示するための定数（LIMIT）を追加します。

以下に、元のコードと、更新後のコードを示します。

変更前:

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

app = Flask(__name__)

変更後:

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__)

pull タスクを追加する（タスクのデータを収集し、pull キューにタスクを作成する）

データモデル Visit は変更されません。fetch_visits() に表示する訪問のクエリも変更されません。コードのこの部分で必要な変更は store_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'
    Visit(visitor='{}: {}'.format(remote_addr, user_agent)).put()

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'
    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)

訪問者トラッキング用のデータモデルとクエリ関数を作成する

データモデル VisitorCount を追加して訪問者をトラッキングします。これには、visitor 自体のフィールドと、訪問数をトラッキングする整数 counter が必要です。次に、fetch_counts() という名前の新しい関数（または Python classmethod）を追加して、上位の訪問者を最も多い順にクエリして返します。fetch_visits() の本体のすぐ下にクラスと関数を追加します。

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

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

ワーカーコードを追加する

/log への GET リクエストで訪問者をログに記録する新しい関数 log_visitors() を追加します。辞書/ハッシュを使用して最新の訪問者数を追跡し、可能な限り多くのタスクを 1 時間リースします。各タスクでは、同じ訪問者によるすべてのアクセスが集計されます。集計結果を基に、アプリは Datastore にすでに存在する対応する VisitorCount エンティティをすべて更新するか、必要に応じて新しいエンティティを作成します。最後のステップでは、処理されたタスクの数と登録されたユーザーの数を示すプレーン テキスト メッセージが返されます。この関数を main.py の fetch_counts() のすぐ下に追加します。

@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))

新しい表示データでメイン ハンドラを更新

上位の訪問者を表示するには、メイン ハンドラ root() を更新して fetch_counts() を呼び出します。また、テンプレートが更新され、上位の訪問者数と最新の訪問数が表示されます。fetch_visits() への呼び出しから訪問者数と最新の訪問をパッケージ化し、単一の context にドロップしてウェブ テンプレートに渡します。以下は、この変更前後のコードです。

変更前:

@app.route('/')
def root():
    'main application (GET) handler'
    store_visit(request.remote_addr, request.user_agent)
    visits = fetch_visits(10)
    return render_template('index.html', visits=visits)

変更後:

@app.route('/')
def root():
    'main application (GET) handler'
    store_visit(request.remote_addr, request.user_agent)
    context = {
        'limit':  LIMIT,
        'visits': fetch_visits(LIMIT),
        'counts': fetch_counts(LIMIT),
    }
    return render_template('index.html', **context)

これらは main.py に必要なすべての変更です。これらの更新を説明するために、main.py に行う変更の概要を示す図を以下に示します。

ad5fd3345efc13d0.png

新しいディスプレイ データでウェブ テンプレートを更新する

ウェブ テンプレート templates/index.html では、最新の訪問者の通常のペイロードに加えて、上位の訪問者を表示するために更新が必要です。上位の訪問者とその数をページ上部の表にドロップし、これまでどおり最新の訪問をレンダリングし続けます。もう 1 つの変更点は、数値をハードコードするのではなく、limit 変数で表示する数値を指定することです。ウェブ テンプレートに加えるべき更新は次のとおりです。

変更前:

<!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>

変更後:

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

<h1>VisitMe example</h1>

<h3>Top {{ limit }} visitors</h3>
<table border=1 cellspacing=0 cellpadding=2>
    <tr><th>Visitor</th><th>Visits</th></tr>
{% for count in counts %}
    <tr><td>{{ count.visitor|e }}</td><td align="center">{{ count.counter }}</td></tr>
{% endfor %}
</table>

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

これで、App Engine Task Queue プルタスクの使用をモジュール 1 サンプルアプリに追加するために必要な変更は完了です。ディレクトリはモジュール 18 サンプルアプリを表すようになり、次のファイルが含まれているはずです。

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

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

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

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

この Codelab の冒頭で gcloud app deploy queue.yaml を使用して行ったように、プルキューがすでに設定されていることを確認します。完了していて、サンプルアプリの準備ができている場合は、gcloud app deploy を使用してアプリをデプロイします。出力は、モジュール 1 のアプリとまったく同じですが、上部に「上位の訪問者」の表が表示されるようになっています。

b667551dcbab1a09.png

更新されたウェブ フロントエンドには上位の訪問者と最新の訪問が表示されますが、訪問者数にはこの訪問は含まれていません。アプリは、以前の訪問者数を表示しながら、この訪問者のカウントを増やす新しいタスクをプルキューにドロップします。このタスクは処理待ちです。

/log を呼び出すことで、さまざまな方法でタスクを実行できます。

たとえば、curl を使用して /log に GET リクエストを送信する場合、PROJECT_ID を指定すると、出力は次のようになります。

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

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

お疲れさまでした。これで、App Engine Task Queue プルキュー サービスの使用をサンプルアプリに追加する Codelab は無事に終了しました。これで、モジュール 19 で Cloud Pub/Sub、Cloud NDB、Python 3 に移行する準備が整いました。

クリーンアップ

全般

現時点で完了した場合は、課金が発生しないように 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 固有のものです。詳細については、各プロダクトのドキュメントをご覧ください。

次のステップ

この「移行」では、訪問者の追跡のサポートを追加することで、モジュール 1 のサンプルアプリにタスクキューの push キューの使用を追加し、モジュール 18 のサンプルアプリを実装しました。次の移行では、App Engine の pull タスクを Cloud Pub/Sub にアップグレードします。2021 年後半以降、Python 3 にアップグレードする際に Cloud Pub/Sub への移行が必須ではなくなりました。詳しくは、次のセクションをご覧ください。

Cloud Pub/Sub への移行については、モジュール 19 の Codelab をご覧ください。それ以外にも、Cloud Datastore、Cloud Memorystore、Cloud Storage、Cloud Tasks（push キュー）など、移行を検討すべきものがあります。Cloud Run と Cloud Functions へのプロダクト間の移行もあります。すべての Serverless Migration Station コンテンツ（Codelab、動画、ソースコード [利用可能な場合]）は、オープンソース リポジトリでアクセスできます。

7. Python 3 への移行

2021 年秋に、App Engine チームは、第 2 世代ランタイム（第 1 世代ランタイムを含む）に対して、多くのバンドル サービスのサポートを拡大しました。その結果、アプリを Python 3 に移植する際に、App Engine タスクキューなどのバンドル サービスから Cloud Pub/Sub などのスタンドアロンの Cloud サービスまたはサードパーティ サービスに移行する必要がなくなりました。つまり、次世代ランタイムからバンドル サービスにアクセスするようにコードを改造すれば、Python 3 App Engine アプリで引き続き Task Queue を使用できます。

バンドル サービスの使用状況を Python 3 に移行する方法については、モジュール 17 の Codelab とその対応する動画をご覧ください。このトピックはモジュール 18 の範囲外ですが、以下に、Python 3 に移植され、App Engine NDB を引き続き使用しているモジュール 1 アプリの Python 3 バージョンへのリンクを示します。（Module 18 アプリの Python 3 バージョンも、今後リリースされる予定です）。

8. 参考情報

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

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

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

移行に関するリソース

モジュール 1（開始）とモジュール 18（終了）のリポジトリ フォルダへのリンクを以下の表に示します。これらのフォルダには、すべての App Engine Codelab 移行のリポジトリからアクセスすることもでき、クローンを作成する、または ZIP ファイルをダウンロードすることができます。

Codelab

Python 2

Python 3

モジュール 1

コード

コード（このチュートリアルでは取り上げません）

モジュール 18（この Codelab）

コード

なし

オンライン リファレンス

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

App Engine のタスクキュー

App Engine プラットフォーム

App Engine のドキュメント

Python 2 App Engine（標準環境）ランタイム

Python 3 App Engine（スタンダード環境）ランタイム

Python 2 と 3 の App Engine（スタンダード環境）ランタイムの違い

Python 2 から 3 への App Engine（スタンダード環境）移行ガイド

App Engine の料金と割り当てに関する情報

第 2 世代の App Engine プラットフォームのリリース（2018 年）

レガシー ランタイムの長期サポート

ドキュメントの移行サンプル

その他のクラウド情報

動画

ライセンス

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