PythonとWerkzeugで作るToDoリストAPI

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)

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

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__': ... はここでは省略

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}')

import json

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

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

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)

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')

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()

from models.todo import Todo

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

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'

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]]