概要
Pythonのflaskフレームワークを使用して、TODOリストを管理するWEBアプリを作成したのでまとめた。
flaskのGET/POSTの方法やセッション等の扱いについて、当アプリを作成しながら振り返ることが目的。
当記事ではログインした利用者に紐づくタスク一覧を表示、タスクのステータス更新について扱う。
概要 Pythonのflaskフレームワークを使用して、TODOリストを管理するWEBアプリを作成したのでまとめた。 flaskのGET/POSTの方法やセッション等の扱いについて、当アプリを作成しながら振り返ることが目的。 当記[…]
タスク一覧表示機能
機能説明
以下の機能をもつTODOトップ画面を実装した。
・ステータスが0(未完了)のタスクを表示する
・タスク一覧は期日の昇順でソートする
・タスクを完了にするボタンを有する
・完了ボタンを押下すると、タスクが完了扱いになりタスク一覧から消える
アプリ概要

前提
タスクを新規登録する機能はまだ作成していないため、予めテーブルにレコードを用意しておく。
◆TODOカテゴリテーブル

◆TODOタスクテーブル

※statusの意味は以下とする。
・1:完了
画面遷移
◆ログイン画面


また、タスクのステータスを完了にするボタンを表示する。
◆【完了】ボタンを押下>エラー有りの場合

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

更新成功の場合、該当のレコードを画面から削除する(テーブルは論理削除)。
全体ファイル構成
ファイル構成は以下とする。

利用者に紐づくタスク一覧表示
サーバー側
モデル修正
タスク一覧はuserテーブルに紐づくtodoテーブルの一覧を表示する。
また、あわせてtodo_categoryテーブルのカテゴリ名称も表示する。
そのため、userテーブルとtodoテーブルとtodo_categoryテーブルを連結した結果を返却する必要がある。
連結したテーブルクラスを定義するため、models.pyは以下のように修正する。
models.py
class Todo(Base):
""" タスククラス """
# テーブル名称
__tablename__ = 'todo'
# カラム
(中略)
category_id = Column(Integer, ForeignKey('todo_category.id'))
user_id = Column(Integer, ForeignKey('user.id'))
# リレーションプロパティ
user = relationship("User", backref="todos")
category = relationship("TodoCategory", backref="todos")
概要 sqlalchemyモジュールを使用してDBを操作する方法についてまとめた。 基本的なCRUD機能の記述方法について扱う。 前提 SQLiteのDB環境を構築していること。
上記により、TodoクラスのオブジェクトをDBから取得した際に、UserクラスとTodoCategoryクラスも参照できるようになる。
※双方向で参照が可能
DBアクセスレイヤ
data_access.py
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
タスク一覧をDBから取得するメソッド。
引数で利用者IDを取得する。
以下の条件でDBから取得したタスク一覧を返却する。
・ステータスが0(未完了)
・タスク期日の昇順でソート
ビジネスロジックレイヤ
business_logic.py
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
data_access.find_todo_all(u_id)メソッドによってTodoクラスのリストが返却されるが、当メソッドにて辞書データのリストに変換する。
ビューレイヤ
views.py
@app.route("/todo_apps/login", methods=["POST"])
def login():
(中略)
# ログイン確認
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:
(略)
ログイン成功時、セッションにユーザーIDを設定するよう修正した。
ログイン後にリダイレクトされるTODOタスク一覧画面表示処理にて、セッションに格納された利用者IDをキーにタスク一覧を検索するよう修正した。
views.py
@app.route("/todo_apps/top")
def top():
""" TODOタスク一覧画面表示 """
# タスク一覧を取得
todos = business_logic.find_todo_all(session["u_id"])
return render_template("top.html", todos=todos)
取得したタスク一覧リスト情報はtodos変数に格納して画面に返却する。
画面側
タスク一覧表示
top.html
<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>
TODOトップ画面ではtodosに格納されたタスク一覧をループして、中の辞書型データを表示している。
タスク完了機能
サーバー側
DBアクセスレイヤ
data_access.py
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
タスクのステータスを1(完了)に更新するメソッド。
t_idをキーにTodoオブジェクトを取得し、ステータスを1(完了)に設定してコミットする。
DB更新に失敗した場合はFalse、更新に成功した場合はTrueを返す。
ビジネスロジックレイヤ
business_logic.py
def update_todo_status(t_id):
""" タスクを完了にする """
return data_access.update_todo_status(t_id)
data_access.pyのステータス更新を呼び出し、結果を返却するメソッド。
ビューレイヤ
views.py
@app.route("/todo_apps/top/task/")
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"))
画面側からt_id(タスクID)を受け取り、対象のタスクのステータスを1(完了)にするメソッド。
更新に失敗した場合、エラーメッセージをerrorsに格納する。
更新に成功した場合、views.py内のtopメソッドへリダイレクトする。
画面側
完了ボタン機能
top.html
<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>
完了ボタンにアンカータグをつけて、views.pyのupdate_todo_doneメソッドへリクエストする。
また、その際にクエリパラメータとしてタスクIDを送る。
完了ボタン押下後エラーの場合
top.html
<div class="card-body">
{% for msg in errors %}
<div id="error-message" class="text-danger mb-3">{{ msg }}</div>
{% endfor %}
ステータス更新時、エラーメッセージが格納された場合はメッセージ内容を表示する。
ファイル詳細
models.py
from sqlalchemy.orm import declarative_base
from sqlalchemy import Column, Integer, String, DateTime, ForeignKey
from sqlalchemy.orm import relationship
# ベースモデル作成
Base = declarative_base()
class User(Base):
""" 利用者クラス """
# テーブル名称
__tablename__ = 'user'
# カラム
id = Column(Integer, primary_key=True)
name = Column(String, nullable=False)
password = Column(String, nullable=False)
class TodoCategory(Base):
""" タスクカテゴリークラス """
# テーブル名称
__tablename__ = 'todo_category'
# カラム
id = Column(Integer, primary_key=True)
category_name = Column(String, nullable=False)
class Todo(Base):
""" タスククラス """
# テーブル名称
__tablename__ = 'todo'
# カラム
id = Column(Integer, primary_key=True)
title = Column(String, nullable=False)
content = Column(String, nullable=False)
memo = Column(String)
status = Column(Integer, default=0)
due_date = Column(DateTime, nullable=False)
category_id = Column(Integer, ForeignKey('todo_category.id'))
user_id = Column(Integer, ForeignKey('user.id'))
# リレーションプロパティ
user = relationship("User", backref="todos")
category = relationship("TodoCategory", backref="todos")
consts.py
# ステータス:未完了 STATUS_INCOMPLETE = 0 # ステータス:完了 STATUS_COMPLETE = 1
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
business_logic.py
import data_access
import consts
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)
views.py
from flask import Flask, render_template, request, redirect, url_for, session
import business_logic
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_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"))
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
@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"))
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('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>

