【Flask】TODOアプリ作成その３：タスク登録機能の実装

概要

Pythonのflaskフレームワークを使用して、TODOリストを管理するWEBアプリを作成したのでまとめた。
flaskのGET/POSTの方法やセッション等の扱い、Jinjaテンプレートの使い方について、当アプリを作成しながら振り返ることが目的。

当記事では利用者に紐づくTODOタスクを新規登録する機能について扱う。

あわせて読みたい

【Flask】TODOアプリ作成その１：ログイン機能の実装ガイド

概要 Pythonのflaskフレームワークを使用して、TODOリストを管理するWEBアプリを作成したのでまとめた。 flaskのGET/POSTの方法やセッション等の扱いについて、当アプリを作成しながら振り返ることが目的。 当記[…]

あわせて読みたい

【Flask】TODOアプリ作成その２：ログイン後のタスク一覧機能の実装

概要 Pythonのflaskフレームワークを使用して、TODOリストを管理するWEBアプリを作成したのでまとめた。 flaskのGET/POSTの方法やセッション等の扱いについて、当アプリを作成しながら振り返ることが目的。 当記[…]

TODOタスクの登録機能

機能説明

以下の機能をもつTODOタスク登録画面を実装した。

アプリ概要

画面遷移

◆TODOトップ画面

◆TODO登録画面

◆【登録】ボタンを押下＞エラー有りの場合

◆【登録】ボタンを押下＞DBエラー有りの場合

DBの更新に失敗した場合、エラーメッセージを表示する。
例えば、todo.dbを更新中であったり、A5SQLを使用してtodo.dbに接続している場合はエラーになる。

入力エラーまたはDB更新エラーが発生した場合、画面入力項目は保持する。

◆【登録】ボタンを押下＞登録成功

タスクの新規登録ができた場合、TODOトップ画面に遷移する。
また、その際には新規登録されたタスクが一覧に表示される。

ファイル構成

ファイル構成は以下とする。
※赤文字箇所は前回からの追加ファイルとなる

解説

カテゴリ一覧表示機能

TODO登録画面では、DBから取得したTODOカテゴリ一覧をセレクトボックスに設定している。

サーバー側

data_access.py

def get_category_all():
    """ カテゴリ一覧取得 """
    session = get_session()
    return session.query(TodoCategory).all()

DBアクセスレイヤでは、TODOカテゴリテーブルのデータをリストで全件取得して返却する。

business_logic.py

def get_category_all():
    """ カテゴリ一覧取得 """
    return data_access.get_category_all()

ビジネスロジックレイヤでは、DBアクセスレイヤで取得したカテゴリ一覧をそのまま返却する。

views.py

@app.route("/todo_apps/create_input")
def show_create():
    """ タスク新規登録画面を表示する """
    categories = business_logic.get_category_all()
    return render_template("create.html", categories=categories)

ルーティングレイヤでは、ビジネスロジックレイヤから取得したカテゴリ一覧をTODO登録画面に返却する。

画面側

create.html

<div class="form-group">
  <label for="category">カテゴリ<span class="text-primary ml-3">※必須</span></label>
  <select class="form-control" id="category" name="category">
    <option value="">選択してください</option>
    {% for category in categories %}
    <option value="{{category.id}}" 
    {% if category.id|string()==request.form.get('category') %} selected {% endif %}>{{category.category_name}}</option>
    {% endfor %}
  </select>
</div>

TODO登録画面のセレクトボックスに、カテゴリ情報を設定する。

最初の初期値は「選択してください」を設定する。
それ以降はサーバー側から取得したカテゴリ一覧を、ループしてセレクトボックスに設定する。

入力チェック機能

TODO登録画面にて入力エラーがある場合、form部品直下にエラーメッセージを表示する。

サーバー側

views.py

@app.route("/todo_apps/create_todo", methods=["POST"])
def create_todo():
    """ タスクを新規登録する """
    # 入力値検証
    error_dict = {}
    is_valid = validate_views.validate_create_todo(request.form, error_dict)
    if not is_valid:
        # 入力エラーの場合
        categories = business_logic.get_category_all()
        return render_template("create.html", categories=categories, error_dict=error_dict)

ルーティングレイヤでは、TODOタスクの登録リクエストがくると画面の入力値検証を行う。
実際の入力値検証処理は、以下のように画面入力チェックレイヤを呼び出してその結果を判定する。

リクエストフォーム内の情報と、入力エラーの場合に画面に返却するエラー辞書情報を引数として渡してる。

入力値検証の結果がエラーの場合、以下のように再度カテゴリ一覧を取得してエラーメッセージと共に返却する。

※カテゴリ一覧はリクエストスコープのため、再度取得しないとセレクトボックスに反映されない。尚、ここではスコープについては扱わない。

validate_views.py

def validate_create_todo(form, error_dict):
    """ タスク新規登録の入力チェックを行う """
    （略）

画面入力チェックレイヤでは、TODO登録画面の入力値検証を行う。

入力エラー項目が存在する場合、is_valid変数にFalseを格納して返却する。
入力エラー項目が存在しない場合は、is_valid変数にTrueを格納して返却する。

必須チェック

# タイトル 必須
title = form.get("title")
if not title:
    error_dict["title"] = "タイトルを入力してください。"
    is_valid = False

必須項目が未入力または未選択の場合、エラー辞書情報にフォーム部品に対応するキーとエラーメッセージを設定する。
また、is_valid変数にFalseを設定する。

セレクトボックスの範囲チェック

# カテゴリ 範囲
categories = business_logic.get_category_all()
id_list = list(map(str, [category.id for category in categories]))
if not c_id in id_list:
    error_dict["category"] = "不正なカテゴリが選択されています。"
    is_valid = False

画面にて選択したセレクトボックスの値が、DBに存在するカテゴリIDに含まれるかチェックを行う。
含まれない場合、画面入力にて改ざんされたということなのでエラーメッセージを設定する。

以下のように記述すると、DBから取得したカテゴリ情報リストをもとに、文字列型のカテゴリIDリストを取得できる。
※内包表記という記述方法で、冗長なコードをシンプルに記述できる

存在しない日付チェック

# タスク期日 存在する日付
try:
    datetime.strptime(due_date, consts.DATE_FORMAT)
except ValueError:
    error_dict["due_date"] = "存在する日付を入力してください。"
    is_valid = False

タスク期日が存在する日付かどうかチェックを行う。
datetimeモジュールを使用して文字列からdatetimeに変換しており、不正な日付文字列の場合は例外が発生する。

例外が発生した場合、エラーメッセージを設定する。

画面側

画面側では入力チェックエラーの場合、エラーメッセージを表示する。
また、リクエストした後で画面入力値がクリアされないよう、入力情報の保持を行う。

create.html

<div class="form-group mt-3">
  <label for="title">タイトル <span class="text-primary ml-3">※必須</span></label>
  <input type="text" class="form-control" id="title" name="title"
    value="{{ request.form.get('title', '') }}">
</div>
{% if error_dict and error_dict["title"] %}
<div id="error-message" class="text-danger mb-3">{{ error_dict["title"] }}</div>
{% endif %}

フォーム部品直下に、以下のようにエラーメッセージ表示領域を設定する。

error_dictを最初に判定する理由は、サーバー側からerror_dictが画面に返却されていない状態で、オブジェクト内を参照しようとするとエラーになってしまうため。
まず、error_dict内に値があるか判定を行う必要がある。

error_dict内に格納されたエラーメッセージは、各フォーム部品の名前をキーに表示する。

・入力内容の保持

セレクトボックス

<select class="form-control" id="category" name="category">
  <option value="">選択してください</option>
  {% for category in categories %}
  <option value="{{category.id}}" 
  {% if category.id|string()==request.form.get('category') %} selected {% endif %}>{{category.category_name}}</option>
  {% endfor %}
</select>

セレクトボックスの場合、入力内容の保持は以下のように記述する。

category.idを文字列に変換した値と、リクエストしたカテゴリの値を比較して等しい場合は選択状態（selected）にする。

テキストボックス

<input type="text" class="form-control" id="title" name="title" value="{{ request.form.get('title', '') }}">

テキストボックスの場合、valueにリクエストした値を設定する。
※リクエストした値を取得できない場合、空文字となる

TODOタスク新規登録機能

TODO登録画面にて入力内容に不備がない場合、TODOタスクテーブルにタスク内容を登録する。
登録に失敗した場合、TODO登録画面にエラーメッセージを表示する。
登録に成功した場合、TODOトップ画面に遷移する。

サーバー側

data_access.py

def insert_todo(todo):
    """ タスクの新規登録 """
    is_valid = True
    
    session = get_session()
    try:
        todo = Todo(
            title = todo["title"],
            content = todo["content"],
            memo = todo["memo"],
            due_date = todo["due_date"],
            category_id = todo["category_id"],
            user_id = todo["user_id"]
        )
        session.add(todo)
        session.commit()
    except Exception as e:
        print(e)
        session.rollback()
        is_valid = False
    finally:
        session.close()

    return is_valid

DBアクセスレイヤでは画面入力値をTodoテーブルにコミットする。

DB登録に失敗した場合、is_valid変数にFalseを格納して返却する。
DB登録に成功した場合、is_valid変数にTrueを格納して返却する。

business_logic.py

def insert_todo(form, u_id):
    """ タスクの新規登録 """
    todo = {
        "title": form.get("title"),
        "content": form.get("content"),
        "memo": form.get("memo", ""),
        "due_date": datetime.strptime(form.get("due_date"), consts.DATE_FORMAT),
        "user_id": u_id,
        "category_id": int(form.get("category"))
    }

    return data_access.insert_todo(todo)

ビジネスロジックレイヤでは、画面入力値をDB登録できる値に変換してDBアクセスレイヤに渡す。
DBアクセスレイヤの結果をそのまま返却する。

 
views.py

# タスク登録
is_valid = business_logic.insert_todo(request.form, session["u_id"])
if not is_valid:
    # 入力エラーの場合
    categories = business_logic.get_category_all()
    error_msg = "タスクの登録に失敗しました。時間を空けて再度実行してください。"
    return render_template("create.html", categories=categories, error_msg=error_msg) 
else:
    return redirect(url_for("top"))

ルーティングレイヤではビジネスロジックレイヤから取得したDB登録結果を判定する。

DB登録に失敗した場合、再度カテゴリ一覧を取得してエラーメッセージと共に返却する。
DB登録に成功した場合、TODOトップ画面表示メソッドにリダイレクトする。

画面側

create.html

<div class="card-body">
  <div id="error-message" class="text-danger mb-3">{{ error_msg }}</div>

DB登録エラーの場合、画面上にエラーメッセージを表示する。

ファイル詳細

consts.py

# ステータス：未完了
STATUS_INCOMPLETE = 0
# ステータス：完了
STATUS_COMPLETE = 1

# 日付変換フォーマット
DATE_FORMAT = "%Y-%m-%d"

data_access.py

from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
from models import User, TodoCategory, Todo
import consts


def get_session():
    """ セッション情報を返却 """
    engine = create_engine('sqlite:///db/todo.db')
    Session = sessionmaker(bind=engine)
    session = Session()

    return session

def find_user(input_name, input_password):
    """ 利用者検索 """
    session = get_session()
    user = session.query(User).filter_by(name=input_name, password=input_password).first()

    return user

def find_todo_all(u_id):
    """ 利用者IDをもとにTODOタスク一覧を取得する """
    session = get_session()
    todo_list = session.query(Todo).filter(Todo.user_id == u_id, Todo.status == consts.STATUS_INCOMPLETE).order_by(Todo.due_date.asc()).all()

    return todo_list

def update_todo_status(t_id):
    """ タスクのステータスを1（完了）に更新する """
    is_valid = True

    session = get_session()
    try:
        todo = session.get(Todo, t_id)
        todo.status = consts.STATUS_COMPLETE
        session.add(todo)
        session.commit()
    except:
        session.rollback()
        is_valid = False
    finally:
        session.close()

    return is_valid

def get_category_all():
    """ カテゴリ一覧取得 """
    session = get_session()
    return session.query(TodoCategory).all()

def insert_todo(todo):
    """ タスクの新規登録 """
    is_valid = True
    
    session = get_session()
    try:
        todo = Todo(
            title = todo["title"],
            content = todo["content"],
            memo = todo["memo"],
            due_date = todo["due_date"],
            category_id = todo["category_id"],
            user_id = todo["user_id"]
        )
        session.add(todo)
        session.commit()
    except Exception as e:
        print(e)
        session.rollback()
        is_valid = False
    finally:
        session.close()

    return is_valid

business_logic.py

import data_access
import consts
from datetime import datetime

def find_user(name, password):
  """ userを取得する """
  # 名前とパスワードからuserオブジェクトを取得
  return data_access.find_user(name, password)

def find_todo_all(u_id):
    """ タスク一覧を取得する """
    # userに紐づくタスク一覧の取得
    todo_list = data_access.find_todo_all(u_id)
    todos = []
    for todo in todo_list:
        # 画面表示用データに整形
        todo_item = {
            "id": todo.id,
            "title": todo.title,
            "category": todo.category.category_name,
            "content": todo.content,
            "due_date": f"{todo.due_date:%Y-%m-%d}",
            "status": "未完了" if todo.status == consts.STATUS_INCOMPLETE else "完了"
        }
        todos.append(todo_item)

    return todos

def update_todo_status(t_id):
    """ タスクを完了にする """
    return data_access.update_todo_status(t_id)

def get_category_all():
    """ カテゴリ一覧取得 """
    return data_access.get_category_all()

def insert_todo(form, u_id):
    """ タスクの新規登録 """
    todo = {
        "title": form.get("title"),
        "content": form.get("content"),
        "memo": form.get("memo", ""),
        "due_date": datetime.strptime(form.get("due_date"), consts.DATE_FORMAT),
        "user_id": u_id,
        "category_id": int(form.get("category"))
    }

    return data_access.insert_todo(todo)

validate_views.py

import business_logic
from datetime import datetime
import consts

def validate_login(form, errors):
    """ ログイン情報の入力チェックを行う """
    is_valid = True
    name = form.get("name")
    password = form.get("password")
    if not name or not password:
        errors.append("名前またはパスワードを入力してください。")
        is_valid = False

    return is_valid

def validate_create_todo(form, error_dict):
    """ タスク新規登録の入力チェックを行う """
    is_valid = True
    # カテゴリ 必須
    c_id = form.get("category")
    if not c_id:
        error_dict["category"] = "カテゴリを選択してください。"
        is_valid = False
    else:
        # カテゴリ 範囲
        categories = business_logic.get_category_all()
        id_list = list(map(str, [category.id for category in categories]))
        if not c_id in id_list:
            error_dict["category"] = "不正なカテゴリが選択されています。"
            is_valid = False
    # タイトル 必須
    title = form.get("title")
    if not title:
        error_dict["title"] = "タイトルを入力してください。"
        is_valid = False
    # タスク内容 必須
    content = form.get("content")
    if not content:
        error_dict["content"] = "タスク内容を入力してください。"
        is_valid = False
    # タスク期日 必須
    due_date = form.get("due_date")
    if not due_date:
        error_dict["due_date"] = "タスク期日を入力してください。"
        is_valid = False
    else:
        # タスク期日 存在する日付
        try:
            datetime.strptime(due_date, consts.DATE_FORMAT)
        except ValueError:
            error_dict["due_date"] = "存在する日付を入力してください。"
            is_valid = False

    return is_valid

views.py

from flask import Flask, render_template, request, redirect, url_for, session
import business_logic
import validate_views 

app = Flask(__name__)
app.secret_key = "yiYKQmFC6MTVKs5THpKkD"

@app.route("/")
def index():
    return redirect(url_for("show_login"))

@app.route("/todo_apps")
def show_login():
    """ ログイン画面表示 """
    return render_template("login.html")

@app.route("/todo_apps/login", methods=["POST"])
def login():
    """ ログイン処理 """
    # 入力値検証
    errors = []
    if not validate_views.validate_login(request.form, errors):
        return render_template("login.html", errors=errors)
    
    # ログイン確認
    name = request.form["name"]
    password = request.form["password"]
    user = business_logic.find_user(name, password)
    if user:
        # ログイン成功
        session["name"] = user.name
        session["u_id"] = user.id
        return redirect(url_for("top"))
    else:
        # ログイン失敗
        errors.append("名前またはパスワードが正しくありません。")
        return render_template("login.html", errors=errors)

@app.route("/todo_apps/top")
def top():
    """ TODOタスク一覧画面表示 """
    # タスク一覧を取得
    todos = business_logic.find_todo_all(session["u_id"])

    return render_template("top.html", todos=todos)

@app.route("/logout")
def logout():
    """ ログアウト """
    session.clear()
    return redirect(url_for("show_login"))

@app.route("/todo_apps/top/task/<t_id>")
def update_todo_done(t_id):
    """ タスクを完了にする """
    is_valid = business_logic.update_todo_status(int(t_id))
    if not is_valid:
        errors = ["更新に失敗しました。時間を空けて再度実行してください。"]
        # タスク一覧を取得
        todos = business_logic.find_todo_all(session["u_id"])

        return render_template("top.html", todos=todos, errors=errors)
    else:
        return redirect(url_for("top"))

@app.route("/todo_apps/create_input")
def show_create():
    """ タスク新規登録画面を表示する """
    categories = business_logic.get_category_all()
    return render_template("create.html", categories=categories)

@app.route("/todo_apps/create_todo", methods=["POST"])
def create_todo():
    """ タスクを新規登録する """
    # 入力値検証
    error_dict = {}
    is_valid = validate_views.validate_create_todo(request.form, error_dict)
    if not is_valid:
        # 入力エラーの場合
        categories = business_logic.get_category_all()
        return render_template("create.html", categories=categories, error_dict=error_dict) 
    else:
        # タスク登録
        is_valid = business_logic.insert_todo(request.form, session["u_id"])
        if not is_valid:
            # 入力エラーの場合
            categories = business_logic.get_category_all()
            error_msg = "タスクの登録に失敗しました。時間を空けて再度実行してください。"
            return render_template("create.html", categories=categories, error_msg=error_msg) 
        else:
            return redirect(url_for("top"))


if __name__ == '__main__':
    # 8080ポートで起動
    app.run(port=8080, debug=True)

top.html

<!DOCTYPE html>
<html>

<head>
  <title>TODOトップ</title>
  <link rel="stylesheet" href="https://stackpath.bootstrapcdn.com/bootstrap/5.0.0-alpha1/css/bootstrap.min.css">
</head>

<body>
  <div class="container mt-5">
    <h2 class="text-center">{{ session["name"] }} さんのタスク一覧</h2>
    <a href="{{ url_for('show_create') }}" class="btn btn-primary mr-3">タスク登録</a>
    <a href="{{ url_for('logout') }}" class="btn btn-primary">ログアウト</a>
    <div class="card mt-4">
      <div class="card-body">
        {% for msg in errors %}
        <div id="error-message" class="text-danger mb-3">{{ msg }}</div>
        {% endfor %}
        <table class="table table-striped">
          <thead>
            <tr>
              <th>カテゴリ</th>
              <th>タイトル</th>
              <th>タスク名</th>
              <th>期日</th>
              <th>ステータス</th>
              <th>アクション</th>
            </tr>
          </thead>
          <tbody>
            {% for todo_item in todos %}
            <tr>
              <td>{{ todo_item["category"] }}</td>
              <td>{{ todo_item["title"] }}</td>
              <td>{{ todo_item["content"] }}</td>
              <td>{{ todo_item["due_date"] }}</td>
              <td>{{ todo_item["status"] }}</td>
              <td><a href="{{ url_for('update_todo_done', t_id=todo_item['id']) }}" class="btn btn-success">完了</a></td>
            </tr>
            {% endfor %}
          </tbody>
        </table>
      </div>
    </div>
  </div>
</body>

</html>

create.html

<!DOCTYPE html>
<html>

<head>
  <title>タスク登録</title>
  <link rel="stylesheet" href="https://stackpath.bootstrapcdn.com/bootstrap/5.0.0-alpha1/css/bootstrap.min.css">
</head>

<body>
  <div class="container">
    <div class="row justify-content-center mt-5">
      <div class="col-md-8">
        <a href="{{ url_for('top') }}" class="btn btn-primary mb-3">戻る</a>
        <div class="card">
          <div class="card-header text-center">
            <h4>タスク登録</h4>
          </div>
          <div class="card-body">
            <div id="error-message" class="text-danger mb-3">{{ error_msg }}</div>
            <form method="post" action="{{ url_for('create_todo') }}">
              <div class="form-group">
                <label for="category">カテゴリ<span class="text-primary ml-3">※必須</span></label>
                <select class="form-control" id="category" name="category">
                  <option value="">選択してください</option>
                  {% for category in categories %}
                  <option value="{{category.id}}" 
                  {% if category.id|string()==request.form.get('category') %} selected {% endif %}>{{category.category_name}}</option>
                  {% endfor %}
                </select>
              </div>
              {% if error_dict and error_dict["category"] %}
              <div id="error-message" class="text-danger mb-3">{{ error_dict["category"] }}</div>
              {% endif %}
              <div class="form-group mt-3">
                <label for="title">タイトル <span class="text-primary ml-3">※必須</span></label>
                <input type="text" class="form-control" id="title" name="title"
                  value="{{ request.form.get('title', '') }}">
              </div>
              {% if error_dict and error_dict["title"] %}
              <div id="error-message" class="text-danger mb-3">{{ error_dict["title"] }}</div>
              {% endif %}
              <div class="form-group mt-3">
                <label for="task">タスク内容 <span class="text-primary ml-3">※必須</span></label>
                <input type="text" class="form-control" id="content" name="content"
                  value="{{ request.form.get('content', '') }}">
              </div>
              {% if error_dict and error_dict["content"] %}
              <div id="error-message" class="text-danger mb-3">{{ error_dict["content"] }}</div>
              {% endif %}
              <div class="form-group mt-3">
                <label for="memo">メモ</label>
                <textarea class="form-control" id="memo" name="memo"
                  rows="3">{{ request.form.get('memo', '') }}</textarea>
              </div>
              <div class="form-group mt-3">
                <label for="due_date">タスク期日<span class="text-primary ml-3">※必須</span></label>
                <input type="date" class="form-control" id="due_date" name="due_date"
                  value="{{ request.form.get('due_date', '') }}">
              </div>
              {% if error_dict and error_dict["due_date"] %}
              <div id="error-message" class="text-danger mb-3">{{ error_dict["due_date"] }}</div>
              {% endif %}
              <button type="submit" class="btn btn-primary mt-3">登録</button>
            </form>
          </div>
        </div>
      </div>
    </div>
  </div>
</body>

</html>