想ひ出19: GCP/Flask1.0/ローカルからDatastoreと接続してみる
おはようございます、moqrinです。
前回に、GCEにデプロイしたFlaskのチュートリアルをCloudSQLと接続しました。
今度はDatastoreを使ってCRUDしてみようぜ、というお話です。
Bookshelf チュートリアルのコードを改悪参考にしています。
なお、基本的に過去の記事からの延長なのでご留意下さい。
やること
1. App Engine アプリケーションを作成
2. サービスアカウントキーを作成して手元に保存
3. 環境変数にサービスアカウントの情報を設定する
4. configにプロジェクトIDを設定
5. モデルを作成
6. Blueprintを修正
7. 画面の修正
8. 確認
1. App Engine アプリケーションを作成
Cloud Datastore API を使用するには、有効化されているApp Engine アプリケーションが必要なので、何か適当に作成します。
引用元: Compute Engine インスタンスから Cloud Datastore API にアクセスする
2. サービスアカウントキーを作成して手元に保存
GCP外のアプリケーションからCloud Datastore API を有効化してアクセスするためには、サービスアカウントキーが必要となります。
Datastoreの読み書きできる権限で作成して、お手元にダウンロードします。
引用元: 別のプラットフォームから Cloud Datastore API にアクセスする
3. 環境変数にサービスアカウントキーを設定する
アクセス許可してもらうためにサービスアカウントキーをOSに設定しておきます。
export GOOGLE_APPLICATION_CREDENTIALS=path/to/ServiceAccountJsonKey
4. configにDatasotreが作成されたプロジェクトIDを設定
手っ取り早くこんな感じで設定しちゃっています。
# instance/config.py PROJECT_ID = 'moqrin-love'
5. モデルを作成
参考: Python Client for Google Cloud Datastore
# app/model_datastore.py from google.cloud import datastore from instance.config import PROJECT_ID import os os.getenv("GOOGLE_APPLICATION_CREDENTIALS") builtin_list = list def init_app(app): pass def get_client(): return datastore.Client(PROJECT_ID) def from_datastore(entity): if not entity: return None if isinstance(entity, builtin_list): entity = entity.pop() entity['id'] = entity.key.id return entity def list(limit=10, cursor=None): ds = get_client() query = ds.query(kind='Blog', order=['-created']) query_iterator = query.fetch(limit=limit, start_cursor=cursor) page = next(query_iterator.pages) entities = builtin_list(map(from_datastore, page)) next_cursor = ( query_iterator.next_page_token.decode('utf-8') if query_iterator.next_page_token else None) return entities, next_cursor def read(id): ds = get_client() key = ds.key('Blog', int(id)) results = ds.get(key) return from_datastore(results) def update(data, id=None): ds = get_client() if id: key = ds.key('Blog', int(id)) else: key = ds.key('Blog') entity = datastore.Entity( key=key, exclude_from_indexes=['description']) entity.update(data) ds.put(entity) return from_datastore(entity) create = update def delete(id): ds = get_client() key = ds.key('Blog', int(id)) ds.delete(key)
6. Blueprintを修正
DatastoreのCRUDを対応します。
# app/blog.py from flask import ( Blueprint, flash, g, redirect, render_template, request, url_for ) from flask_login import current_user, login_required from datetime import datetime from . import model_datastore bp = Blueprint('blog', __name__) @bp.route('/', methods=['GET', 'POST']) @login_required def index(): token = request.args.get('page_token', None) if token: token = token.encode('utf-8') posts, next_page_token = model_datastore.list(cursor=token) return render_template('blog/index.html', posts=posts, next_page_token=next_page_token) @bp.route('/create', methods=('GET', 'POST')) @login_required def create(): if request.method == 'POST': title = request.form['title'] body = request.form['body'] error = None if not title: error = 'Title is required.' if error is not None: flash(error) else: post = {'username': current_user.username, 'title': title, 'body': body, 'author_id': g.user.id, 'created': datetime.now().strftime("%Y/%m/%d %H:%M:%S") } model_datastore.create(post) flash('Successfully added a new post.') return redirect(url_for('blog.index')) return render_template('blog/create.html') @bp.route('/<int:id>/update', methods=('GET', 'POST')) @login_required def update(id): post = model_datastore.read(id) if request.method == 'POST': title = request.form['title'] body = request.form['body'] error = None if not title: error = 'Title is required.' if error is not None: flash(error) else: post = { 'username': current_user.username, 'title': title, 'body': body, 'author_id': g.user.id, 'created': datetime.now().strftime("%Y/%m/%d %H:%M:%S") } model_datastore.create(post, id) flash('Successfully edited the post.') return redirect(url_for('blog.index')) return render_template('blog/update.html', post=post) @bp.route('/<int:id>/delete', methods=('POST',)) @login_required def delete(id): model_datastore.delete(id) flash('Successfully deleted the post.') return redirect(url_for('blog.index'))
7. 画面の修正
修正する画面はindexだけです。
# app/templates/blog/index.html
{% extends 'base.html' %}
{% block header %}
<h1>{% block title %}Posts{% endblock %}</h1>
{% if current_user.is_authenticated %}
<a class="action" href="{{ url_for('blog.create') }}">New</a>
{% endif %}
{% endblock %}
{% block content %}
{% for post in posts %}
<article class="post">
{% if current_user.username == post.username %}
<header>
<div>
<h1>{{ post.title }}</h1>
</div>
<a class="action" href="{{ url_for('blog.update', id=post.id) }}">Edit</a>
<div class="about">by {{ post.username }} on {{ post.created }}</div>
</header>
<p class="body">{{ post.body }}</p>
</article>
{% if not loop.last %}
{% endif %}
<hr>
{% endif %}
{% endfor %}
{% endblock %}
8. 確認
Google CloudのAPI Client libraryをインストール
pip install gcloud
起動。
export FLASK_APP=run export FLASK_ENV=development flask run
表示を確認して喜びます。
↓画面↓
↓Datastore↓
わーい。 でも...何か動作おせーー。。。ｗｗ
参考:
Python での Cloud Datastore の使用
Python Client for Google Cloud Datastore
Datastoreに外部からアクセスする方法
