PythonでFlaskを使ったTodoアプリの構築

このチュートリアルでは、Todo アプリの API (ウェブサービス) を構築する予定です。API サービスは、REST ベースのアーキテクチャを使用して実装されます。

このアプリは、次のような主な機能を備えています。

• ToDoリストへのアイテムの作成
• ToDoリスト全体を読み込む
• 未着手」、「進行中」、「完了」のいずれかのステータスを持つ項目を更新する。
• リストからのアイテムの削除

RESTとは？

REST (REpresentational State Transfer) は、WebサービスやAPIを構築するためのアーキテクチャスタイルの1つです。RESTを実装するシステムは、ステートレスであることが要求されます。クライアントは、サーバーがどのような状態であるかを知ることなく、リソースを取得または変更するためにサーバーにリクエストを送信します。サーバーは、クライアントとの直前の通信が何であったかを知る必要なく、クライアントに応答を送信する。

RESTfulシステムへの各リクエストは、一般的に以下の4つのHTTP動詞を使用します。

• GET。GET：特定のリソースまたはリソースの集合を取得する
• GET：特定のリソースまたはリソースのコレクションを取得します。新しいリソースを作成する
• PUT。特定のリソースを更新する
• DELETE: 特定のリソースを削除する

PATCH、HEAD、OPTIONSのような他のものも許可され、時々使用されますが。

Flaskとは？

FlaskはWebアプリケーションを開発するためのPythonのフレームワークです。Flask は非主観的なフレームワークで、あなたに代わって判断することはありません。そのため、特定の方法でアプリケーションを構成することを制限することはありません。そのため、Flaskを使用する開発者には、より大きな柔軟性と制御性が提供されます。Flask は Web アプリを作成するための基本的なツールを提供し、アプリに含める必要があるほとんどのものを簡単に拡張することができます。

Flask の代替として、他の有名な Web フレームワークも検討することができます。Djangoは、Flaskがうまくいかない場合の最も人気のある代替品の1つです。このチュートリアルでは、Django と Flask の比較を行いました。

Flaskのセットアップ

まず、pipでFlaskをインストールします。

$ pip install Flask

早速、Flask を設定して、ローカルマシンに Web サーバを立ち上げてみましょう。todo_service_flaskディレクトリにmain.py` というファイルを作成します。

from flask import Flask
app = Flask(__name__)


@app.route('/')
def hello_world():
    return 'Hello World!'

Flask をインポートした後、ルートを設定します。ルートは、URLパターン、HTTPメソッド、HTTPリクエストを受け取って処理する関数で指定します。そのルートに、そのURLがHTTPでリクエストされるたびに呼び出されるPython関数をバインドしています。この例では、ルート（/）のルートを設定し、URLパターン http://[IP-OR-DOMAIN]:[PORT]/ でアクセスできるようにしました。

Flaskアプリを実行する

次の仕事は、ローカルサーバーを立ち上げてこのWebサービスを提供し、クライアントからアクセスできるようにすることです。

ありがたいことに、これはすべて1つのシンプルなコマンドで行うことができます。

$ FLASK_APP=main.py flask run

コンソールにメッセージが表示されるはずです。

Running on http://127.0.0.1:5000/ (Press CTRL+C to quit)

cURLを使用して、GETリクエストを発行することができます。Macを使用している場合、cURLはすでにシステムにインストールされているはずです。

$ curl -X GET http://127.0.0.1:5000/

レスポンスが返ってくるはずです。

Hello World!

話はここで終わりではありません。Todoアプリケーションを構成していきましょう。

Todoアプリの構造化

Todo アプリは、いくつかの基本的な機能を備えています。

• リストへのアイテムの追加
• リストからの全アイテムの取得
• リスト内のアイテムを更新する
  リスト内のアイテムの更新 * リスト内のアイテムの削除

これらは、create、read、update、delete を意味する CRUD 操作と呼ばれることがあります。

データの保存には、非常に軽量なファイルベースのデータベースであるSQLiteを使用します。DB Browser for SQLiteをインストールすれば、簡単にデータベースを作成することができる。

このデータベースを todo.db という名前にして、ディレクトリ todo_service_flask の下に置いてみましょう。さて、テーブルを作成するために、単純にクエリを実行します。

CREATE TABLE "items" (
    "item" TEXT NOT NULL,
    "status" TEXT NOT NULL,
    PRIMARY KEY("item")
);

また、シンプルにするために、すべてのルートを1つのファイルに書きますが、特に非常に大きなアプリでは、これは必ずしも良い習慣ではありません。

また、ヘルパー関数を格納するために、もう一つのファイルを使用します。これらの関数には、データベースに接続して適切なクエリを実行することで、リクエストを処理するビジネスロジックが含まれます。

この最初の Flask 構造に慣れたら、好きなようにアプリを再構築できます。

アプリの構築

データベースへのアイテムの追加など、よく実行されるタスクで何度もロジックを書くのを避けるために、別のファイルにヘルパー関数を定義して、必要なときに呼び出すようにします。このチュートリアルでは、このファイルを helper.py と名付けます。

アイテムの追加

この機能を実装するために、2つのものが必要です。

• データベースに新しい要素を追加するビジネスロジックを含むヘルパー関数
• 特定の HTTP エンドポイントにヒットしたときに呼び出されるルート

まず、いくつかの定数を定義して、 add_to_list() 関数を書いてみましょう。

import sqlite3


DB_PATH = './todo.db'   # Update this path accordingly
NOTSTARTED = 'Not Started'
INPROGRESS = 'In Progress'
COMPLETED = 'Completed'


def add_to_list(item):
    try:
        conn = sqlite3.connect(DB_PATH)


# Once a connection has been established, we use the cursor
        # object to execute queries
        c = conn.cursor()


# Keep the initial status as Not Started
        c.execute('insert into items(item, status) values(?,?)', (item, NOTSTARTED))


# We commit to save the change
        conn.commit()
        return {"item": item, "status": NOTSTARTED}
    except Exception as e:
        print('Error: ', e)
        return None

この関数はデータベースとの接続を確立し、挿入クエリを実行します。挿入された項目とそのステータスを返します。

次に、いくつかのモジュールをインポートして、パス /item/new のルートをセットアップします。

import helper
from flask import Flask, request, Response
import json


app = Flask(__name__)


@app.route('/')
def hello_world():
    return 'Hello World!'


@app.route('/item/new', methods=['POST'])
def add_item():
    # Get item from the POST body
    req_data = request.get_json()
    item = req_data['item']


# Add item to the list
    res_data = helper.add_to_list(item)


# Return error if item not added
    if res_data is None:
        response = Response("{'error': 'Item not added - " + item + "'}", status=400 , mimetype='application/json')
        return response


# Return response
    response = Response(json.dumps(res_data), mimetype='application/json')


return response

requestモジュールはリクエストを解析し、HTTP のボディデータやクエリパラメータを URL から取得するために使用されます。response はクライアントにレスポンスを返すために使用される。レスポンスは JSON 型である。

PythonでJSONを読み書きする方法についてもっと読みたい方は、こちらをご覧ください。

クライアントのエラーによりアイテムが追加されなかった場合、ステータス400を返しました。json.dumps()`関数は、Pythonのオブジェクトや辞書を有効なJSONオブジェクトに変換してくれます。

コードを保存して、私たちの機能が正しく実装されているかどうかを検証してみましょう。

cURLを使用してPOSTリクエストを送信し、アプリをテストすることができます。また、POSTボディとしてアイテム名を渡す必要があります。

$ curl -X POST http://127.0.0.1:5000/item -d '{"item": "Setting up Flask"}' -H 'Content-Type: application/json'

Windows を使用している場合、JSON データをシングルクォートからダブルクォートにフォーマットし、エスケープする必要があります。

$ curl -X POST http://127.0.0.1:5000/item -d "{"item": "Setting up Flask"}" -H 'Content-Type: application/json'

以下の点に注意してください。

• ベースURL(http://127.0.0.1:5000)とルートまたはパス(/item/new)です。
• リクエストメソッドは POST です。
• リクエストがウェブサーバーに到達すると、この情報に基づいてエンドポイントの位置を特定しようとします。
• データはJSON形式で渡しています – {“item”: “Flaskのセットアップ”}

リクエストを実行すると、レスポンスが返ってくるはずです。

{"Setting up Flask": "Not Started"}

次のコマンドを実行して、リストに1つアイテムを追加してみましょう。

$ curl -X POST http://127.0.0.1:5000/item -d '{"item": "Implement POST endpoint"}' -H 'Content-Type: application/json'

タスクの説明とステータスが表示されたレスポンスが返ってくるはずです。

{"Implement POST endpoint": "Not Started"}

おめでとうございます! これで、Todoリストに項目を追加する機能が実装できました。

全項目の取得

リストからすべての項目を取得したいことはよくありますが、これはありがたいことにとても簡単です。

def get_all_items():
    try:
        conn = sqlite3.connect(DB_PATH)
        c = conn.cursor()
        c.execute('select * from items')
        rows = c.fetchall()
        return { "count": len(rows), "items": rows }
    except Exception as e:
        print('Error: ', e)
        return None

この関数はデータベースとの接続を確立し、SELECTクエリを作成して c.fetchall() を介して実行します。これは、SELECTクエリによって返されたすべてのレコードを返します。もし、1つの項目だけに興味があるのなら、代わりに c.fetchone() を呼び出すことができます。

このメソッド get_all_items は、2つの項目を含む Python オブジェクトを返します。

• このクエリによって返されたアイテムの数
• クエリによって返された実際のアイテム

main.py で、GET リクエストを受け付けるルート /item/new を定義します。ここで、 @app.route() に methods キーワード引数を渡しません。なぜなら、この引数を省略すると、デフォルトで GET になってしまうからです。

@app.route('/items/all')
def get_all_items():
    # Get items from the helper
    res_data = helper.get_all_items()


# Return response
    response = Response(json.dumps(res_data), mimetype='application/json')
    return response

cURLを使用してアイテムを取得し、ルートをテストしてみましょう。

$ curl -X GET http://127.0.0.1:5000/items/all

レスポンスが返ってくるはずです。

json {"count": 2, "items": [["Setting up Flask", "Not Started"], [Implement POST endpoint", "Not Started"]]}.

個別アイテムの状態取得

前の例と同様に、このためのヘルパー関数を書きます。

def get_item(item):
try:
    conn = sqlite3.connect(DB_PATH)
    c = conn.cursor()
    c.execute("select status from items where item='%s'" % item)
    status = c.fetchone()[0]
    return status
except Exception as e:
    print('Error: ', e)
    return None

また、main.pyにリクエストを解析し、レスポンスを返すルートを定義します。ルートはGETリクエストを受け付ける必要があり、アイテム名はクエリパラメータとして送信される必要があります。

クエリパラメータは、URLと一緒に ?name=value というフォーマットで渡されます。例えば、http://base-url/path/to/resource/?name=value のようになります。もし、値の中にスペースがある場合は、+ か %20 (スペースをURLエンコードしたもの) に置き換える必要がある。名前と値のペアを & 文字で区切ると、複数の名前と値のペアを持つことができます。

以下は、クエリーパラメータの有効な例です。

• http://127.0.0.1:8080/search?query=what+is+flask
• http://127.0.0.1:8080/search?category=mobiles&brand=apple

@app.route('/item/status', methods=['GET'])
def get_item():
    # Get parameter from the URL
    item_name = request.args.get('name')


# Get items from the helper
    status = helper.get_item(item_name)


# Return 404 if item not found
    if status is None:
        response = Response("{'error': 'Item Not Found - %s'}"  % item_name, status=404 , mimetype='application/json')
        return response


# Return status
    res_data = {
        'status': status
    }


response = Response(json.dumps(res_data), status=200, mimetype='application/json')
    return response

もう一度、cURLを使用してリクエストを発行してみましょう。

$ curl -X GET http://127.0.0.1:5000/item/status?name=Setting+up+Flask

すると、次のようなレスポンスが返ってくるはずです。

{"status": "Not Started"}

項目の更新

Flask のセットアップ」のタスクは少し前に完了したので、そろそろステータスを「完了」に更新しましょう。

まずは、更新クエリを実行する関数を helper.py に書いてみましょう。

def update_status(item, status):
    # Check if the passed status is a valid value
    if (status.lower().strip() == 'not started'):
        status = NOTSTARTED
    elif (status.lower().strip() == 'in progress'):
        status = INPROGRESS
    elif (status.lower().strip() == 'completed'):
        status = COMPLETED
    else:
        print("Invalid Status: " + status)
        return None


try:
        conn = sqlite3.connect(DB_PATH)
        c = conn.cursor()
        c.execute('update items set status=? where item=?', (status, item))
        conn.commit()
        return {item: status}
    except Exception as e:
        print('Error: ', e)
        return None

エンドユーザーが私たちのアプリケーションで何をするか分からないので、ユーザーの入力に頼らず、バリデーションを行うことは良い習慣です。ここでは非常に単純な検証を行っていますが、これが実際のアプリケーションであれば、SQL インジェクション攻撃のような他の悪意のある入力から保護したいと思うでしょう。

次に、リソースを更新するためのPUTメソッドを受け付けるルートを main.py で設定します。

@app.route('/item/update', methods=['PUT'])
def update_status():
    # Get item from the POST body
    req_data = request.get_json()
    item = req_data['item']
    status = req_data['status']


# Update item in the list
    res_data = helper.update_status(item, status)


# Return error if the status could not be updated
    if res_data is None:
        response = Response("{'error': 'Error updating item - '" + item + ", " + status   +  "}", status=400 , mimetype='application/json')
        return response


# Return response
    response = Response(json.dumps(res_data), mimetype='application/json')


return response

先ほどと同じように、cURLを使用してこのルートをテストしてみましょう。

$ curl -X PUT http://127.0.0.1:5000/item/update -d '{"item": "Setting up Flask", "status": "Completed"}' -H 'Content-Type: application/json'

次のようなレスポンスが返ってくるはずです。

{"Setting up Flask": "Completed"}

アイテムの削除

まず最初に、削除クエリを実行する関数を helper.py に記述します。

def delete_item(item):
    try:
        conn = sqlite3.connect(DB_PATH)
        c = conn.cursor()
        c.execute('delete from items where item=?', (item,))
        conn.commit()
        return {'item': item}
    except Exception as e:
        print('Error: ', e)
        return None

Note: (item,) はタイプミスではないことに注意してください。タプルの中にアイテムが1つしかない場合でも、 execute() にタプルを渡す必要があります。カンマを追加することで、強制的にタプルになります。

次に、DELETEリクエストを受け付けるルートを main.py で設定します。

@app.route('/item/remove', methods=['DELETE'])
def delete_item():
    # Get item from the POST body
    req_data = request.get_json()
    item = req_data['item']


# Delete item from the list
    res_data = helper.delete_item(item)


# Return error if the item could not be deleted
    if res_data is None:
        response = Response("{'error': 'Error deleting item - '" + item +  "}", status=400 , mimetype='application/json')
        return response


# Return response
    response = Response(json.dumps(res_data), mimetype='application/json')


return response

cURLを使用して、この削除ルートをテストしてみましょう。

$ curl -X DELETE http://127.0.0.1:5000/item/remove -d '{"item": "Setting up Flask"}' -H 'Content-Type: application/json'

レスポンスが返ってくるはずです。

{"item": "Temporary item to be deleted"}

これで、必要なバックエンドの機能をすべて備えたアプリが完成しました!

結論

このチュートリアルで、Flask を使って簡単な REST ベースの Web アプリケーションを作る方法を理解してもらえたと思います。もしあなたがDjangoのような他のPythonフレームワークの経験があるなら、Flaskを使う方がずっと簡単だと観察したかもしれません。

このチュートリアルでは、GUIを使わないバックエンドアプリケーションの側面に焦点を当てましたが、Flaskを使ってHTMLページやテンプレートをレンダリングすることもできますが、それは別の記事で紹介することにします。

Flaskを使ってHTMLテンプレートを管理するのは全く問題ありませんが、ほとんどの人はFlaskを使ってバックエンドのサービスを構築し、アプリのフロントエンド部分は人気のあるJavaScriptライブラリのどれかを使って構築しています。自分に一番合うものを試してみてください。あなたのFlaskの旅に幸あれ

もしソースコードで遊んでみたい、または上記のコードから実行するのが難しい場合は、こちらがGitHubにあります!