PythonとWerkzeugで作るToDoリストAPI

シンプルなToDoリストのWeb APIを作る。前回はWSGIの仕様のみを参考にして作ったが、今回はWerkzeugというライブラリを利用する。

Werkzeugとは

Werkzeugのドキュメントの “Werkzeug is a comprehensive WSGI web application library.“という文面にもある通り、これはWSGIのWebアプリを作るためのライブラリである。あくまでフレームワークではなく、ライブラリであることに注意する。 Webアプリを作るうえで、便利なクラスや関数が用意された「道具箱」のようなイメージを持つとよいかもしれない (そもそも"werkzeug"という単語はドイツ語で「道具」という意味)。あくまで道具があるだけなので、どういう設計を行うかなどを考える余地がある。

ToDoリストAPIの仕様

前回と同じだが、再掲する。

簡単のため、今回ToDoのデータはidと内容のみ持つデータとし、{ id: 0, "content": "やること" }というJSON形式とする。

APIの仕様は以下の通り。

URL	Method	説明	返却値
`/todo/`	GET	全てのToDoを取得。	ToDoのデータのリスト
`/todo/`	POST	ToDoを作成。	なし (LocationヘッダにそのToDoへのURLを乗せる)
`/todo/<todo_id>`	GET	`todo_id`のidを持つToDoを取得。	ToDoのデータ
`/todo/<todo_id>`	PUT	`todo_id`のidを持つToDoを変更。	なし
`/todo/<todo_id>`	DELETE	`todo_id`のidを持つToDoを削除	なし

データは最終的にはSQLiteで管理する。

雛形

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

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
from werkzeug.wrappers import Response


def app(env, start_response):
    response = Response('Hello, World')
    return response(env, start_response)


if __name__ == '__main__':
    from werkzeug.serving import run_simple
    run_simple('127.0.0.1', 5000, app, use_debugger=True, use_reloader=True)

以降、サーバーを起動する際はpython3 app.pyを実行する。

試しに、コマンドラインで接続を確認してみると良い。

% curl localhost:5000
Hello, World

雛形 - 解説

Responseを使わないWSGIアプリは以下のようになる。

1
2
3
def app(env, start_response):
    start_response('200 OK', [('Content-type', 'text/plain; charset=utf-8')])
    return [b'Hello, World.']

一方でResponseを使えば、レスポンスに関する情報をResponseに包むことができるため、コードの意味が分かりやすくなっている。 Responseのインスタンスは__call__(env, start_response)を実装しており、これもまたWSGIアプリとみなせる。それゆえ、app関数の最後でreturn response(env, start_response)のような使い方をしている。

Responsオブジェクトについてはドキュメントを読めばわかるが、引数にステータスコード、ヘッダ、MIMEタイプなどを指定することができる (ドキュメントには明記されていないが、mimetypeを未指定にすると勝手にtext/plain、エンコーディングはUTF8になるようだ)。

werkzeug.servingではデバッグ用のWSGIサーバーが用意されており、run_simpleとして利用できる。詳細はドキュメントを参照。 use_debugger引数をTrueにすると、サーバー側でデバッグ用のログが出力されるようになる。 use_reloader引数をTrueにすると、ファイルの変更を検知しサーバーを自動でリロードしてくれる。

ルーティング

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
31
32
33
34
35
36
37
38
from wsgiref.simple_server import make_server
from werkzeug.wrappers import Request, Response
from werkzeug.routing import Map, Rule, Submount
from werkzeug.exceptions import HTTPException
from views import todo


def hello(request: Request):
    return Response('Hello, World')


URL_MAP = Map([
    Rule('/', endpoint=hello),
    Submount('/todo', [
        Rule('/', methods=['GET'], endpoint=todo.get_all),
        Rule('/', methods=['POST'], endpoint=todo.post),
        Rule('/<int:todo_id>/', methods=['GET'], endpoint=todo.get),
        Rule('/<int:todo_id>/', methods=['PUT'], endpoint=todo.put),
        Rule('/<int:todo_id>/', methods=['DELETE'], endpoint=todo.delete)
    ])
])


def route(request: Request):
    adapter = URL_MAP.bind_to_environ(request.environ)
    try:
        endpoint, values = adapter.match()
        return endpoint(request, **values)
    except HTTPException as e:
        return e


def app(env, start_response):
    request = Request(env)
    response = route(request)
    return response(env, start_response)

# if __name__ == '__main__': ... はここでは省略

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

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
from werkzeug.wrappers import Request, Response


def get_all(request: Request):
    return Response('todo.get_all')


def post(request: Request):
    return Response('todo.post')


def get(request: Request, todo_id: int):
    return Response(f'todo.get: {todo_id}')


def put(request: Request, todo_id: int):
    return Response(f'todo.put: {todo_id}')


def delete(request: Request, todo_id: int):
    return Response(f'todo.delete: {todo_id}')

リクエストを試しに送ってみる。

% curl localhost:5000/todo/
todo.get_all
% curl -X POST localhost:5000/todo/
todo.post
% curl localhost:5000/todo/1/
todo.get: 1
% curl -X PUT localhost:5000/todo/1/
todo.put: 1
% curl -X DELETE localhost:5000/todo/1/
todo.delete: 1

ルーティング - 解説

まずapp関数に注目する。Requestは、envを使いやすいように包んだクラスである。詳細はRequestのドキュメントを参照。

例えば、Requestを使わずにenvを使ってjsonデータを取得するには、以下のようにしなければならない。

1
2
3
4
5
import json

content_length = int(env['CONTENT_LENGTH']) # 中身が空かどうかの処理はここではサボっている
body_str = env['wsgi.input'].read(content_length)
body_json = json.loads(body_str)

Responseを使うと、これが簡潔になる。なお、jsonのデコードに失敗した場合は、デフォルトでBadRequest例外を投げる (詳細はget_jsonのドキュメント参照)。

1
2
# response = Response(env)とした
body_json = response.get_json()

app関数では、envをRequestで包んだ後、ルーティングの処理をroute関数に任せている。

前回は、「どのURLに対してどの関数が呼ばれるか」という対応関係を、listとtupleのみで表現していた。 werkzeug.routingでは、これをMapとRuleで管理する。Submountを使うと、/todoをprefixに持つURLをまとめることができる。

上の例のように、/<int:todo_id>/とすると、URLに含まれる整数値をtodo_idとして取得することができる。intの部分はconverterと呼ばれ、標準で使えるconverterはドキュメントで確認できる。前回の正規表現を使った取り出し方よりも見やすくなっている。

Mapの使い方はroute関数を見るとわかる。URL_MAP.bind_to_environメソッドでenvironとURL_MAPを結びつけ、 adapter.matchメソッドで実際にURL、メソッドの照合を行う。返却値は対応するendpointと、URLに含まれる変数である。

URL_MAPの中に対応するURL、メソッドが存在しなかった場合は、NotFound例外を投げる。これはHTTPExceptionを継承したクラスである。前回はこれらを自前で実装したが、werkzeugではwerkzeug.exceptionsで実装されている。

ToDoリストのインターフェース作成

前回同様、以下のTodoクラスを作る。

メソッド(引数)	説明
`__init__(self, content)`	`content`を内容とするToDoを作成。
`insert(self)`	`todo`をToDoリストに追加。
`update(self)`	`todo`の変更を反映させる。
`delete(self)`	idが`todo_id`であるToDoを削除

クラスメソッド(引数)	説明	例外
`get_all(cls)`	全てのToDoを取得。
`get(cls, todo_id)`	idが`todo_id`であるToDoを取得。	`todo_id`が存在しなかった場合に`TodoNotFound`例外を投げる。

上の定義で、実際にviews/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
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
from werkzeug.wrappers import Request, Response
from models.todo import Todo, TodoNotFound
from werkzeug.exceptions import BadRequest, NotFound
import json


def todo_to_dict(todo):
    return {'id': todo.id, 'content': todo.content}


class ValidationError(Exception):
    pass


def validate_todo(todo_dict):
    if 'content' not in todo_dict:
        raise ValidationError

    if type(todo_dict['content']) is not str:
        raise ValidationError


def get_all(request: Request):
    todos_dict = [todo_to_dict(todo) for todo in Todo.get_all()]
    todos_json = json.dumps(todos_dict, ensure_ascii=False)
    return Response(todos_json, mimetype='application/json')


def post(request: Request):
    todo_dict = request.get_json(force=True)

    try:
        validate_todo(todo_dict)
    except ValidationError:
        raise BadRequest('Todo ValidationError')

    todo_id = Todo(todo_dict['content']).insert()
    return Response('', status=201, headers=[('Location', f'/todo/{todo_id}/')])


def get(request: Request, todo_id: int):
    try:
        todo = Todo.get(todo_id)
    except TodoNotFound:
        raise NotFound('Todo NotFound')
    todo_dict = todo_to_dict(todo)
    todo_json = json.dumps(todo_dict, ensure_ascii=False)
    return Response(todo_json, mimetype='application/json')


def put(request: Request, todo_id: int):
    todo_dict = request.get_json(force=True)

    try:
        validate_todo(todo_dict)
        todo = Todo.get(todo_id)
    except ValidationError:
        raise BadRequest('Todo ValidationError')
    except TodoNotFound:
        raise NotFound('Todo NotFound')

    todo.content = todo_dict['content']
    todo.update()
    return Response('', status=204)


def delete(request: Request, todo_id: int):
    try:
        todo = Todo.get(todo_id)
    except TodoNotFound:
        raise NotFound('Todo NotFound')
    todo.delete()
    return Response('', status=204)

ひとまずダミーを実装する。models/__init__.pyと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
class TodoNotFound(Exception):
    pass


class Todo:
    def __init__(self, content: str):
        self.id = None
        self.content = content

    def insert(self):
        self.id = 0
        return self.id

    def update(self):
        pass

    def delete(self):
        pass

    @classmethod
    def get_all(cls):
        return []

    @classmethod
    def get(cls, todo_id: int):
        return Todo('dummy')

ToDoリストのインターフェース作成 - 解説

todo_to_dictやvalidate_todoの処理は前回と全く同じ。

get_allやgetなどの関数の引数はRequestクラスのインスタンスである。また、関数の返却値はResponseのオブジェクトを返すようにしている。これは、app.pyのapp関数やroute関数の実装による。

Requestクラスのインスタンスであることで、JSONの取得にget_jsonメソッドが使える。 get_jsonメソッドの引数にforce=Trueを指定すると、リクエストヘッダのMIMEタイプを気にしないようにできる。今回それをやるのは大した理由ではなく、単にcurlでテストするときにMIMEタイプを指定するのが面倒だから(あまり行儀は良くないと思うが…)。

NotFoundやBadRequestなどのクラスを投げる際に、クラスそのものではなくNotFound(message)のようにインスタンスを投げることができる。すると、レスポンスにmessageが出力される。以下は、次節のmodels/todo.pyのコードのうえで実行した例(上のコードではこのレスポンスは返ってこないことに注意)。

% curl localhost:5000/4/
<!DOCTYPE HTML PUBLIC "-//W3C//DTD HTML 3.2 Final//EN">
<title>404 Not Found</title>
<h1>Not Found</h1>
<p>Todo NotFound</p>

ToDoリストの実装

ここでは、SQLiteでデータを管理する。内容は前回と全く同じ。 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
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
import sqlite3


class TodoNotFound(Exception):
    pass


class Todo:
    TODOS = []
    NEXT_ID = 0
    DB_PATH = 'db.sqlite3'

    def __init__(self, content: str, todo_id=None):
        self.id = todo_id
        self.content = content

    def insert(self):
        with Todo.get_db() as con:
            con.execute('insert into Todo(content) values (?)', (self.content,))
            cur = con.cursor()
            row = cur.execute('select * from Todo where id=last_insert_rowid()').fetchone()
            self.id = row['id']
        return self.id

    def update(self):
        if self.id is None:
            raise TodoNotFound
        with Todo.get_db() as con:
            con.execute('update Todo set content=? where id=?', (self.content, self.id))

    def delete(self):
        if self.id is None:
            raise TodoNotFound
        with Todo.get_db() as con:
            con.execute('delete from Todo where id=?', (self.id,))

    @classmethod
    def get_all(cls):
        with Todo.get_db() as con:
            cur = con.cursor()
            rows = cur.execute('select * from todo').fetchall()
        return [Todo(row['content'], row['id']) for row in rows]

    @classmethod
    def get(cls, todo_id: int):
        with Todo.get_db() as con:
            cur = con.cursor()
            row = cur.execute('select * from todo where id=?', (todo_id,)).fetchone()
            if not row:
                raise TodoNotFound
        return Todo(row['content'], row['id'])

    @classmethod
    def get_db(cls):
        con = sqlite3.connect(Todo.DB_PATH)
        con.row_factory = sqlite3.Row
        return con

    @classmethod
    def init_table(cls):
        with Todo.get_db() as con:
            con.execute('drop table if exists Todo')
            con.execute('''create table Todo(
                              id integer primary key autoincrement,
                              content text not null
                           )''')
            con.commit()

データベースを初期化するためにはTodo.init_table()を実行する。そこで、初期化用のスクリプトinit_db.pyを作成する。

1
2
3
4
from models.todo import Todo

if __name__ == '__main__':
    Todo.init_table()

データベースを初期化したい時は、コマンドラインで以下のように実行する。

% python3 init_db.py

うまく動いているか確認する。

% curl -X POST localhost:5000/todo/ -d '{"content": "部屋の掃除"}'

% curl -X POST localhost:5000/todo/ -d '{"content": "犬の散歩"}'

% curl -X POST localhost:5000/todo/ -d '{"content": "風呂を洗う"}'

% curl localhost:5000/todo/
[{"id": 1, "content": "部屋の掃除"}, {"id": 2, "content": "犬の散歩"}, {"id": 3, "content": "風呂を洗う"}]%

% curl localhost:5000/todo/2/
{"id": 2, "content": "犬の散歩"}

% curl -X PUT localhost:5000/todo/1/ -d '{"content": "aaa"}'

% curl localhost:5000/todo/
[{"id": 1, "content": "部屋の掃除"}, {"id": 2, "content": "aaa"}, {"id": 3, "content": "風呂を洗う"}]%

% curl -X DELETE localhost:5000/todo/2/

% curl localhost:5000/todo/
[{"id": 1, "content": "部屋の掃除"}, {"id": 3, "content": "風呂を洗う"}]

werkzeug.testとpytestを用いたテスト

curlでうまく動いているのか確認するだけでなく、ちゃんとテストコードを書いて検証してみる。今回はテストのツールとしてpytestを用いる。また、werkzeug.testモジュールのClientを使えば、サーバーを起動せずにクライアントを作成することができ、リクエストの送受信をシミュレーションできる。

pytestはあまり使った経験がないため、pytestについても備忘のため簡単に解説を書く。

簡単な使い方

tests/test_todo.pyを作成して、ひとまず内容を次のようにする。

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
import pytest
from models.todo import Todo
import tempfile
import os
from werkzeug.test import Client
from app import app


@pytest.fixture
def client():
    db_fd, Todo.DB_PATH = tempfile.mkstemp()
    Todo.init_table()
    yield Client(app)

    os.close(db_fd)
    os.unlink(Todo.DB_PATH)


def test_hello(client):
    res = client.get('/')
    assert res.get_data(as_text=True) == 'Hello, World'

テストを実行する場合、以下のコマンドを実行する。

% pytest

うまくいくと1 passedというメッセージが出力される。

簡単な使い方 - 解説

pytestコマンドを実行すると、testsディレクトリのtest_*.pyというファイル内のtest_*という関数を実行してテストを行う。よって、上の例ではtests/test_todo.pyのtest_helloがテストの対象となる。

@pytest.fixtureというデコレータを定義すると、テストの前処理を行うことができる。この前処理をfixtureという。デコレータで修飾された関数の返却値は、テスト関数の引数として利用できるようになる。ただし、その引数名はpytest.fixtureで修飾した関数と同名にする必要がある。上の例では、client関数をpytest.fixtureで修飾しているから、test_hello関数ではclient関数の返却値を引数clientとして使えるようになる。

補足: ここでは行っていないが、conftest.pyにfixtureを書いて、他のテストファイルから呼び出すことができる。

Clientはwerkzeug.testで定義されているクラス。引数にWSGIアプリを与えてインスタンス化することで、クライアントを作成することができる。このインスタンスを使えば、GETやPOSTなどのリクエストをgetやpostメソッドで行える。実際、test_helloではclient.get('/')でURL '/'にGETリクエストを送っている。 Clientの詳細はドキュメントを参照。

Clientオブジェクトのgetやpostメソッドの返却値はTestResponseクラスのインスタンスで、get_dataやget_jsonなどのメソッドでデータを取得できる。詳細はドキュメントを参照。

pytestにおいて、結果の正しさを確認するためには、シンプルにassert文を使う。上の例では、レスポンスで得られたデータres.get_data()が'Hello, World'であるかどうかを判定している。

client関数では、Clientクラスのインスタンスを作るだけでなく、データベースの初期化を行っている。また、そのテスト限りのデータベースを作成するために、その一時ファイルをtempfile.mkstemp()関数で作成している。これはファイルのディスクリプタとファイルのパスをタプルで返すため、後者をTodoのデータベースのパスに設定している。 os.closeとos.unlinkを使って、テストが終わった後に一時ファイルを削除している。

Todoのテスト

大体のことは上で説明したので、後はテストを書くだけ。 tests/test_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
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
import pytest
from models.todo import Todo
import tempfile
import os
from werkzeug.test import Client
from app import app
import json
from urllib.parse import urlparse


@pytest.fixture
def client():
    db_fd, Todo.DB_PATH = tempfile.mkstemp()
    Todo.init_table()
    yield Client(app)

    os.close(db_fd)
    os.unlink(Todo.DB_PATH)


@pytest.fixture
def todos_req():
    todo1 = {'id': 1, 'content': '部屋の掃除'}
    todo2 = {'id': 2, 'content': '犬の散歩'}
    todo3 = {'id': 3, 'content': 'ご飯の準備'}
    return [todo1, todo2, todo3]


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


def test_add_todos(client, todos_req):
    for todo_req in todos_req:
        res = client.post('/todo/', data=json.dumps({"content": todo_req['content']}))
        assert res.status_code == 201

        todo_id = todo_req['id']
        loc_path = urlparse(res.headers.get('Location')).path
        assert loc_path == f'/todo/{todo_id}/'

    todos_res = client.get('/todo/').get_json()
    assert todos_req == todos_res

    todo_res = client.get('/todo/1/').get_json()
    assert todos_req[0] == todo_res


def test_put_todo(client, todos_req):
    for todo_req in todos_req:
        res = client.post('/todo/', data=json.dumps({"content": todo_req['content']}))

    todo_new = {'id': 2, 'content': 'aaa'}
    res = client.put('/todo/2/', data=json.dumps({"content": todo_new['content']}))
    assert res.status_code == 204

    res = client.get('/todo/')
    assert res.get_json() == [todos_req[0], todo_new, todos_req[2]]


def test_delete_todo(client, todos_req):
    for todo_req in todos_req:
        res = client.post('/todo/', data=json.dumps({"content": todo_req['content']}))

    res = client.delete('/todo/2/')
    assert res.status_code == 204

    res = client.get('/todo/')
    assert res.get_json() == [todos_req[0], todos_req[2]]

pytestコマンドを実行すると、4 passedというメッセージが出力される。

ソースコード

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

Werkzeugとは#

ToDoリストAPIの仕様#

雛形#

雛形 - 解説#

ルーティング#

ルーティング - 解説#

ToDoリストのインターフェース作成#

ToDoリストのインターフェース作成 - 解説#

ToDoリストの実装#

werkzeug.testとpytestを用いたテスト#

簡単な使い方#

簡単な使い方 - 解説#

Todoのテスト#

ソースコード#