PythonとFlask(+α)で作るToDoリストAPI

シンプルなToDoリストのWeb APIを作る。 今までWSGIの仕様のみ、Werkzeug の2通りで実装したが、今回はFlaskといくつかのライブラリを使う。使うのは以下の通り。

• Flask: WSGIアプリフレームワーク。
• peewee: ORMライブラリ。
• marshmallow: データの変換やvalidationをするためのライブラリ。

Flaskとは

WSGIのWebアプリを作るためのフレームワーク。 フレームワークであるため、Flaskが用意した作法に従ってコードを書くことで比較的手軽にWebアプリが作成できる。 前回との比較でいうならば、例えばルーティングの仕組みをプログラマが書く必要はない。これはFlaskに備わっている。

Djangoのようなフレームワークとは違って、持っている機能は少ない。必要に応じて外部ライブラリを組み合わせる。 例えば、Djangoではデフォルトでデータベースの仕組みが内蔵されているが、Flaskにはない。その代わりに、 データベースのライブラリとしてsqlite3やSQLAlchemy、peeweeなど、好きなものを用いれば良い。

ToDoリストAPIの仕様

今回ToDoのデータは以下キーと値を持つJSONデータとする。

APIの仕様は以下の通り。

データはSQLiteで管理する。

雛形

todo_listディレクトリを作成。todo_list/__init__.pyを以下のようにする。

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11

from flask import Flask


def create_app():
    app = Flask(__name__)

    @app.route('/')
    def hello():
        return 'Hello, World'

    return app

exportコマンドで環境変数を設定する。

% export FLASK_APP=todo_list
% export FLASK_ENV=development

flask runでサーバーが起動する。

% flask run

curlコマンドで試しにアクセスしてみる。

% curl localhost:5000
Hello, World

雛形 - 解説

まず、Flask(__name__)でWSGIアプリを作ることができる。アプリをappとしたとき、@app.routeデコレータでURLに対応する関数を登録することができる。 以下は、URL /に対して関数helloを登録している。よって、URL /にアクセスされたときに、Hello, Worldというレスポンスが返ってくるようになる。

1
2
3

    @app.route('/')
    def hello():
        return 'Hello, World'

HTMLデータやJSONなどの、アプリのユーザーが「見る部分」のことをviewという。 そしてhelloのような、@app.routeで登録される関数をview関数という。 上のコードでは'Hello, World'を返すviewをview関数helloとして定義していることになる。 view関数の書き方についてはドキュメントのQuickstartを参照。 例えば文字列ではなくdictを返すと、JSONデータとしてレスポンスが送られる。

create_app関数はWSGIアプリを返す関数である。この関数内で、WSGIアプリに関する様々な設定をする。 このようなパターンはApplication Factoriesと呼ばれる。 後でここにデータベースのパスの設定など、色々書いていく。

環境変数FLASK_APPには、WSGIアプリがあるモジュールを指定している。 flask runコマンドは、FLASK_APPに設定されたモジュールのcreate_app関数を元にWSGIアプリを作成し、開発用サーバーを起動する。 環境変数FLASK_ENVにdevelopmentを設定すると、自動リロードがONになったり、例外発生時にデバッグ用のレスポンスを送ったりしてくれる。

Blueprintの利用

まず、Blueprintを作成する。todo_list/todo.pyを作成し、ひとまず内容を以下の通りにする。

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30

from flask import Blueprint, request

bp = Blueprint('todo', __name__)


@bp.route('/', methods=["GET"])
def get_all():
    return 'get_all'


@bp.route('/', methods=["POST"])
def post():
    body = request.get_json()
    return f'post: {body}'


@bp.route('/<int:todo_id>/', methods=["GET"])
def get(todo_id: int):
    return f'get: {todo_id}'


@bp.route('/<int:todo_id>/', methods=["PUT"])
def put(todo_id: int):
    body = request.get_json()
    return f'put: {todo_id}, {body}'


@bp.route('/<int:todo_id>/', methods=["DELETE"])
def delete(todo_id: int):
    return f'delete: {todo_id}'

WSGIアプリにBlueprintを登録する。todo_list/__init__.pyの内容を以下の通りにする。

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15

from flask import Flask


def create_app():
    app = Flask(__name__, url_prefix='/todo')

    @app.route('/')
    def hello():
        return 'Hello, World'

    # 以下2文を追加
    from . import todo
    app.register_blueprint(todo.bp)

    return app

curlで確認。

% curl localhost:5000/todo/
get_all
% curl localhost:5000/todo/1/
get: 1
% curl -X POST -H "Content-Type: application/json" localhost:5000/todo/ -d '{"content": "部屋の掃除"}'
post: {'content': '部屋の掃除'}
% curl -X PUT -H "Content-Type: application/json" localhost:5000/todo/1/ -d '{"content": "部屋の掃除"}'
put: 1, {'content': '部屋の掃除'}
% curl -X DELETE localhost:5000/todo/1/
delete: 1

Blueprint - 解説

Blueprintとは、viewをグループ化する仕組みである。 これによって、viewの使い回しができたり、同じprefixのURLでまとめたりできるようになる (参考)。

上のコードでは、todoというBlueprintを作り、それをbpとした。 使い方は@app.routeと同じく、@bp.routeのように用いる。 WSGIアプリにBlueprintを登録するには、register_blueprintメソッドを使う。 url_prefix引数を/todoとすることで、対応するBlueprintのURLのprefixを/todoに統一する。

peeweeを使ったToDoモデルの作成

todo_list/models/__init__.pyとtodo_list/models/todo.pyを作成。前者は空のままにし、後者は内容を以下のようにする。

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12

from peewee import Model, TextField, DateTimeField
from datetime import datetime


class Todo(Model):
    content = TextField()
    created_at = DateTimeField(default=datetime.now)
    updated_at = DateTimeField()

    def save(self, *args, **kwargs):
        self.updated_at = datetime.now()
        return super(Todo, self).save(*args, **kwargs)

peeweeを使ったToDoモデルの作成 - 解説

peeweeはORMである。つまり、PythonのオブジェクトとRDBMS(上ではSQLite)のデータを相互に変換するためのライブラリである。 ORMを使う1つの利点は、SQL文を書かずに済むことである。 データはModelというクラスとして管理する。Modelを介してRDBMSのデータを操作する。

上のコードではTodoというモデルを作成し、その属性としてcontent, created_at, updated_at を用意している。 これらはそれぞれ、テキスト、時刻、時刻の型を持つ。idは自動で追加される。

created_atには、Todoのインスタンス作成時に時刻が記録されるように、default引数を設定している。 updated_atについて、saveメソッドが呼ばれる度に時刻が記録されるように、save()メソッドをオーバーライドしている (参考)。

peeweeを使ったToDoモデルの作成 - コンソール上でのテスト

うまくTodoが定義されているかテストしてみる。

まず、関連のモジュールをimportする。

% python3
>>> from todo_list.models.todo import Todo
>>> from peewee import SqliteDatabase

次に、データベースを作成する。テストのため、データベースはメモリ上で動かす。 データベースにTodoクラスを紐づけた後、create_tableでTodoのテーブルを作成する。

>>> db = SqliteDatabase(':memory:')
>>> db.bind([Todo])
>>> db.create_table([Todo])

実はこのあたりはpeeweeのQuickstartとやり方が異なる。 Quickstartでは、TodoクラスのMeta内部クラスにデータベースを設定している。 ではなぜ今回はbindを使ったのかと言うと、Flaskのアプリを作る上での都合である。詳しくは後述する。 bindを使った話はSetting the database at run-timeに載っている。

Todo.createでTodoのインスタンスを作成。selectやgetなどのクラスメソッドで、データの取得が行える。saveメソッドでデータの更新ができる。

>>> todo = Todo.create(content='部屋の掃除')
>>> list(Todo.select())
[<Todo: 1>]
>>> todo.content = '犬の散歩'
>>> todo.save()
1
>>> Todo.get(Todo.id == 1).content
'犬の散歩'

peeweeでは便利な機能をplayhouseとして提供しているようで、 例えばplayhouse.shortcutsのmodel_to_dict関数は、Modelをdictに変換してくれる。

>>> from playhouse.shortcuts import model_to_dict
>>> model_to_dict(todo)
{'id': 1, 'content': '犬の散歩', 'created_at': datetime.datetime(2021, 6, 2, 15, 45, 15, 247438), 'updated_at': datetime.datetime(2021, 6, 2, 15, 45, 39, 839677)}

ただし、dictへの変換はmarshmallowに任せるので、model_to_dictはこの後使わない。

marshmallowを使ったデータの変換とvalidation

todo_list/models/todo.pyを以下のようにする。

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19

from peewee import Model, TextField, DateTimeField
from datetime import datetime
from marshmallow import Schema, fields


class Todo(Model):
  # ...
  # 省略
  # ...


class TodoSchema(Schema):
    id = fields.Integer()
    content = fields.Str(required=True)
    created_at = fields.DateTime()
    updated_at = fields.DateTime()


todo_schema = TodoSchema()

marshmallowを使ったデータの変換とvalidation - 解説

marshmallowは、JSONデータとPythonオブジェクトを相互に変換する機能を提供する。 その変換のルールはSchemaクラスで定義する。 上のTodoSchemaは、以下の変換のルールを定義する。

• idは整数値
• contentは文字列。必須項目なので、required=Trueを設定している。
• create_at、update_atは日付時刻。
  DateTimeは時刻を文字列として変換する。その形式のデフォルトはISO8601(参考)。

スキーマのインスタンスをtodo_schemaとする。todo_schema.loadsでJSONからdict、todo_schema.dumpsでPythonオブジェクトからJSONに変換する。 変換の際に、フィールドが存在しなかったり、型が合わなかったりした場合は、ValidationError例外を投げる。

marshmallowを使ったデータの変換とvalidation - コンソール上でのテスト

再びPythonでテストする。

まずはデータベースを作る。

>>> from todo_list.models.todo import Todo
>>> from peewee import SqliteDatabase
>>> db = SqliteDatabase(':memory:')
>>> db.bind([Todo])
>>> db.create_table([Todo])

Todoからdictへの変換はdumpで行う。JSONに変換するdumpsというのもある。

>>> todo = Todo.create(content='部屋の掃除')
>>> from todo_list.models.todo import todo_schema
>>> todo_schema.dump(todo)
{'created_at': '2021-06-02T16:32:55.820672', 'updated_at': '2021-06-02T16:32:55.820739', 'content': '部屋の掃除', 'id': 1}'

dictのvalidationはloadで行う。不正な値をloadするとValidationErrorが発生することも確認する。

>>> todo_schema.load({'content': '部屋の掃除'})
{'content': '部屋の掃除'}
>>> todo_schema.load({'foo': 1})
...
marshmallow.exceptions.ValidationError: {'content': ['Missing data for required field.'], 'foo': ['Unknown field.']}
>>> todo_schema.load({'content': 1})
...
marshmallow.exceptions.ValidationError: {'content': ['Not a valid string.']}

Flaskとデータベースの連携

todo_list/db.pyを作成し、内容を以下のようにする。

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16

from peewee import SqliteDatabase
from .models.todo import Todo
from flask import Flask

db = SqliteDatabase(None)


def init_app(app: Flask):
    db.init(app.config['DB_PATH'])
    db.bind([Todo])
    app.teardown_appcontext(close_db)


def close_db(e=None):
    if not db.is_closed():
        db.close()

todo_list/app.pyを以下のようにする。

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30

from flask import Flask
import os # 追加


# 追加
def make_instance_path(app):
    try:
        os.makedirs(app.instance_path)
    except OSError:
        pass


def create_app():
    app = Flask(__name__)
    app.config['DB_PATH'] = os.path.join(app.instance_path, 'db.sqlite3') # 追加

    make_instance_path(app) # 追加

    # 以下の2行を追加
    from . import db
    db.init_app(app)

    @app.route('/')
    def hello():
        return 'Hello, World'

    from . import todo
    app.register_blueprint(todo.bp, url_prefix="/todo")

    return app

todo_list/todo.pyに以下の内容を追記する。

1
2
3
4
5

from .db import db

@bp.before_request
def connect_db():
    db.connect()

Flaskとデータベースの連携 - 解説

Flaskでは、WSGIアプリの設定情報を管理するためにapp.configが提供されている。 app.configの設定方法はいくつかある。 外部のPythonファイルから読み込んだり(app.config.from_pyfile)、 dictから読み込んだり(app.config.from_mapping)できる。 これをうまく利用すれば、例えば本番環境と開発環境、テスト環境で異なる設定を適用することができる。 ここでは、データベースのパスをapp.config['DB_PATH']に格納しておく。

Flaskにはinstance folderというものがある。 これは、例えばGitでバージョン管理してほしくないもの(データベースファイルや設定ファイルなど)を入れるディレクトリである。 そのパスはapp.instance_pathで取得でき、デフォルトでは「アプリケーションの1つ上のディレクトリ」となる。 今回の場合はtodo_listと同じ階層あるinstanceというディレクトリをinstance folderとみなす。 instance folderは自動で作られることはないため、その作成をmake_instance_folderに任せている。

データベースの設定はdb.init_app関数に任せる。 db.init_appでは、まずデータベースの設定をdb.initで行う。その後db.bindでTodoを紐付けている。 SQLiteDatabaseの引数は本来データベースのパスを指定するはずだが、これをNoneとすることで、それを後回しにしている。 そして、そのパスをdb.init(app.config['DB_PATH'])で設定している (参考)。 前前節で「TodoクラスのMeta内部クラスにデータベースを設定しないのはFlaskアプリを作るうえでの都合」と述べたが、 その理由はこのようにデータベースのパスをapp.configで設定したかったからだ。

app.teardown_appcontextでは、リクエストが送られてきて、それに対して何かしらの処理が終わった後に呼び出される関数を登録する。 ここでは、データベースがもし接続されていたら閉じる処理を行う。

/todo/...のURLにアクセスされた時は必ずにデータベースに接続するので、その接続処理をconnect_dbで行う。@bp.before_requestデコレータを使うと、 リクエストの前に行われる処理を設定することができる。

teardown_appcontextやbefore_requestで、リクエスト処理の前後に新たな処理を付け足せることが見て取れるだろう。 こういう機能が提供されていることはありがたい。

データベースの初期化

todo_list/db.pyを以下のようにする。

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33

from peewee import SqliteDatabase
from .models.todo import Todo
from flask import Flask
# 以下2行を追加
from flask.cli import with_appcontext
import click

db = SqliteDatabase(None)


def init_app(app: Flask):
    db.init(app.config['DB_PATH'])
    db.bind([Todo])
    app.teardown_appcontext(close_db)
    app.cli.add_command(init_db_command) # 追加


def close_db(e=None):
    if db.is_closed():
        db.close()

# 追加
def init_db():
    db.drop_tables([Todo])
    db.create_tables([Todo])


# 追加
@click.command('init-db')
@with_appcontext
def init_db_command():
    init_db()
    click.echo('Initizlized the database.')

以下のコマンドを実行すると、データベースが初期化される。

% flask init-db
Initizlized the database.

データベースの初期化 - 解説

Flaskでは自作のコマンドを作る機能が備わっている。コマンドに対応する関数を登録するメソッドはapp.cli.add_commandで行う。 登録する関数に対して@click.commandデコレータを使うと、コマンド名を設定できる。上のコードではその名前をinit-dbにしているため、 flask init-dbでコマンドが実行される。@with_appcontextで包むと、WSGIアプリが生成された後にこのコマンドが実行されることを保証してくれる。 今回の例ではinit_appが先に実行されていないとデータベースに接続できないため、このデコレータを用いている。

コマンドの内容は、単にTodoのテーブルを削除して、新たに作り直しているだけである。 drop_tablesでは、SQLのDROP文が実行される。これはデフォルトでIF EXISTS句をつけてくれる。この設定はsafe引数で変えられる。 同様に、create_tablesではデフォルトでIF NOT EXISTSをつけてくれる。

ToDoのレスポンスの実装

todo_list/todo.pyを以下のようにする。

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45

from flask import Blueprint, request, jsonify
from .db import db
from .models.todo import Todo, dump_todo, load_todo_or_400

bp = Blueprint('todo', __name__)


@bp.before_request
def connect_db():
    db.connect()


@bp.route('/', methods=["GET"])
def get_all():
    todos = list(Todo.select())
    return jsonify(dump_todo(todos, many=True))


@bp.route('/', methods=["POST"])
def post():
    todo_dict = load_todo_or_400(request.get_json())
    todo = Todo.create(content=todo_dict['content'])
    return '', 201, [('Location', f'/todo/{todo.id}/')]


@bp.route('/<int:todo_id>/', methods=["GET"])
def get(todo_id: int):
    todo = Todo.get_or_404(todo_id)
    return jsonify(dump_todo(todo))


@bp.route('/<int:todo_id>/', methods=["PUT"])
def put(todo_id: int):
    todo = Todo.get_or_404(todo_id)
    todo_dict = load_todo_or_400(request.get_json())
    todo.content = todo_dict['content']
    todo.save()
    return '', 204


@bp.route('/<int:todo_id>/', methods=["DELETE"])
def delete(todo_id: int):
    todo = Todo.get_or_404(todo_id)
    todo.delete_instance()
    return '', 204

todo_list/models/todo.pyを以下のようにする。

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34

from peewee import Model, TextField, DateTimeField, DoesNotExist
from datetime import datetime
from marshmallow import Schema, fields, ValidationError
from werkzeug.exceptions import NotFound, BadRequest


class Todo(Model):
    # ... 省略 ...

    @classmethod
    def get_or_404(cls, todo_id: int):
        try:
            todo = Todo.get(Todo.id == todo_id)
        except DoesNotExist:
            raise NotFound
        return todo


class TodoSchema(Schema):
  # ... 省略 ...


todo_schema = TodoSchema()


def dump_todo(obj, many=False):
    return todo_schema.dump(obj, many=many)


def load_todo_or_400(obj):
    try:
        return todo_schema.load(obj)
    except ValidationError:
        raise BadRequest

todo_list/__init__.pyに1行追加する。

1
2
3
4
5

def create_app():
    app = Flask(__name__)
    app.config['DB_PATH'] = os.path.join(app.instance_path, 'db.sqlite3')
    app.config['JSON_AS_ASCII'] = False # 追加
    # ...省略

ToDoのレスポンスの実装 - 解説

レスポンスボディを取得したり、Todoを返してデータベースを操作したり、それをJSONに変換して送ったりしているだけ。

getではToDo、get_allでは、ToDoのリストを返す必要がある。そのために、dump_todo関数を定義している。 many引数で、変換後の値がリストか否かを制御する。

get, put, deleteでは、idがtodo_idに一致するToDoを探す処理を行っている。もし見つからなければ404 NotFoundを返すように、 クラスメソッドget_or_404を定義している。

レスポンスボディが正しい形式であるかを判定するために、load_todo_or_400を定義している。 もし正しい形式でなければ、400 BadRequestを返す。

レスポンスで日本語がエスケープされないようにするためには、app.config['JSON_AS_ASCII']をFalseに設定する。 レスポンスとしてjsonify関数を返すと、MIMEタイプをapplication/jsonに設定してくれる。 なので、todo_schema.dumpsを使わずにあえてjsonifyを使っている。

pytestによるテスト

Werkzeugと同様、Flaskにもテスト用のクライアントが用意されている。これについて軽く書いておく。

pytestによるテスト - 準備

まず、アプリ側でテスト用の設定を受け付けられるようにする。todo_list/__init__.pyのcreate_app関数を以下のように追記する。

 1
 2
 3
 4
 5
 6
 7
 8
 9
10

def create_app(test_config: dict = None): # 追加
    app = Flask(__name__)
    app.config['DB_PATH'] = os.path.join(app.instance_path, 'db.sqlite3')
    app.config['JSON_AS_ASCII'] = False

    # 以下2行を追加
    if test_config is not None:
        app.config.from_mapping(test_config)

    # ... 省略 ...

引数にtest_configが設定されていた場合、app.config.from_mappingメソッドでそれをapp.configに割り当てる。

testsディレクトリを作成し、tests/conftest.pyを作成。内容を以下のようにする。 これはFlaskのTutorialに書いてある内容とほぼ同じである。

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33

import pytest
from todo_list import create_app
from todo_list.db import init_db
import tempfile
import os


@pytest.fixture
def app():
    db_fd, db_path = tempfile.mkstemp()

    app = create_app({
        'TESTING': True,
        'DB_PATH': db_path,
    })

    with app.app_context():
        init_db()

    yield app

    os.close(db_fd)
    os.unlink(db_path)


@pytest.fixture
def client(app):
    return app.test_client()


@pytest.fixture
def runner(app):
    return app.test_cli_runner

appのfixtureをまず作成し、それを使ってclientとrunnerのfixtureを作成する。 clientはテスト用のクライアントを返す。これはもちろんアプリのテストに用いる。 runnerはコマンドのテストを行うために用いる。

テスト用のアプリでは、データベースを一時ファイルとして作成するため、その初期処理をapp fixtureで行っている。

pytestによるテスト - コマンド

tests/test_db.pyを作成し、内容を以下のようにする。 これはFlaskのTutorialとほぼ同じ。

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11

def test_init_db(runner, monkeypatch):
    class Recorder:
        called = False

    def fake_init_db():
        Recorder.called = True

    monkeypatch.setattr('todo_list.db.init_db', fake_init_db)
    result = runner.invoke(args=['init-db'])
    assert 'Initialized' in result.output
    assert Recorder.called

runner.invokeでコマンドを実行し、その出力結果をresult.outputで取得できる。 ちゃんとInitializedで始まるメッセージが出力されているかをテストしている。

monkeypatchは、pytestで標準で用意されているfixtureである。 init_dbが呼ばれたかどうかをテストするために、monkeypatchを使ってtodo_init_.db.init_dbをfake_init_dbにすり替え、 fake_init_dbが呼ばれたかどうかのフラグをRecorderクラスで管理している。

pytestによるテスト - ToDo

tests/test_todo.pyを作成し、内容を以下のようにする。 本当はいろいろテストしなければならないのだが、ここでは一例のみ示す。

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11

def test_get_all(client):
    res = client.get('/todo/')
    assert res.get_json() == []

    contents = ["aaa", "bbb", "ccc"]
    for content in contents:
        res = client.post('/todo/', json={'content': content})

    res = client.get('/todo/')
    for content, todo in zip(contents, res.get_json()):
        assert content == todo['content']

clientはWerkzeugのものと使い方は同じ。 例えばclient.getやclient.postで、それぞれGETメソッド、POSTメソッドを送ることができる。

ソースコード

ソースコードはGitHubのRepositryにあげた。