想ひ出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↓

わーい。
でも…何か動作おせーー。。。ww

参考:

Python での Cloud Datastore の使用
Python Client for Google Cloud Datastore
Datastoreに外部からアクセスする方法

シェアする

  • このエントリーをはてなブックマークに追加

フォローする

%d人のブロガーが「いいね」をつけました。