ぼんの備忘録

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

Flask
目次

概要

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

当記事ではログインした利用者に紐づくタスク一覧を表示、タスクのステータス更新について扱う。

 

あわせて読みたい
【Flask】TODOアプリ作成その１：ログイン機能の実装ガイド

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

 

タスク一覧表示機能

機能説明

以下の機能をもつTODOトップ画面を実装した。

・ログインした利用者に紐づくタスク一覧を表示する
・ステータスが0（未完了）のタスクを表示する
・タスク一覧は期日の昇順でソートする
・タスクを完了にするボタンを有する
・完了ボタンを押下すると、タスクが完了扱いになりタスク一覧から消える

 

アプリ概要

 

前提

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

◆TODOカテゴリテーブル

◆TODOタスクテーブル

※statusの意味は以下とする。

・0：未完了
・1：完了

画面遷移

◆ログイン画面

 

◆TODOトップ画面
未完了のタスク一覧を期日の昇順でソートする。
また、タスクのステータスを完了にするボタンを表示する。

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

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ではjoinまたはrelationshipを利用してテーブル結合が可能。
あわせて読みたい
【sqlalchemy】PythonでDB操作：SQLAlchemyの基本的な使用方法

概要 sqlalchemyモジュールを使用してDBを操作する方法についてまとめた。 基本的なCRUD機能の記述方法について扱う。   前提 SQLiteのDB環境を構築していること。

今回はrelationshipを使用してuser、todo、todo_categoryテーブルを連結した結果を返却している。

上記により、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から取得したタスク一覧を返却する。

・利用者ID（ログインしている）に紐づく
・ステータスが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
タスク一覧を画面表示しやすいように整形するメソッド。
引数で利用者IDを取得する。
data_access.find_todo_all(u_id)メソッドによってTodoクラスのリストが返却されるが、当メソッドにて辞書データのリストに変換する。
また、その際に期日をyyyy-mm-dd形式に変換し、ステータスは意味がわかりやすいよう文字列に変換する。
ビューレイヤ

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>

 

 

 

 

スポンサーリンク