[Flask-SQLAlchemy]レコード更新ログを実装する/autoincrementの値をレコード挿入時に取得する方法

2020年12月8日

Contents

1. はじめに
2. 前回まで
3. ER図を確認
4. モデル作成
5. 処理の実装
6. 編集時、データに変更がなければ編集しない処理
7. 動作確認
8. まとめ
9. 参考

はじめに

PythonのWebアプリケーションフレームワークFlaskを使って、「消耗品買い物リスト」を作ってみるシリーズ第13回目です。

今回は、新規登録/更新/削除処理が行われる際に、レコードの更新ログを残す処理を追加します。

前回まで

ER図を確認

改めてER図を確認します。
最初に書いたものは、Itemの状態(active/delete)を確認するstatusカラムが抜けていたので追加しました。

Itemレコードに変更が発生した場合、そのレコードをそのままItemUpdateLogレコードにする、という処理にします。

モデル作成

ItemモデルとItemUpdateLogモデルにstatusを追加して反映します。

tbl/models/models.py

(略)
class Item(db.Model):
    """消耗品・アイテムのモデル
    """
    __tablename__ = 'items'
    __table_args__ = {'extend_existing': True}

    id = db.Column(db.Integer, primary_key=True, autoincrement=True)
    user_page_id = db.Column(db.Integer, db.ForeignKey('user_pages.id', onupdate='CASCADE', ondelete='CASCADE'))
    item_category_id = db.Column(db.Integer, db.ForeignKey('item_categories.id', onupdate='CASCADE', ondelete='CASCADE'))
    name = db.Column(db.String(255), nullable=False)
    purchase_cycle_days = db.Column(db.Integer, default=30) # 購入頻度
    stock_rate = db.Column(db.Float, default=1.0) # 現在の在庫割合
    status = db.Column(db.String(255), nullable=False, default="active") # アイテムの利用ステータス。初期値はactive。削除するとdeleted

    created_at = db.Column(db.DateTime, nullable=False, default=datetime.now)
    updated_at = db.Column(db.DateTime, nullable=False, default=datetime.now, onupdate=datetime.now)
(略)
class ItemUpdateLog(db.Model):
    """アイテムの変更ログ。すべての値を記録しておく
    """
    __tablename__ = 'item_update_logs'
    __table_args__ = {'extend_existing': True}

    id = db.Column(db.Integer, primary_key=True, autoincrement=True)
    item_id = db.Column(db.Integer, db.ForeignKey('items.id', onupdate='CASCADE', ondelete='CASCADE'))
    item_category_id = db.Column(db.Integer, db.ForeignKey('item_categories.id', onupdate='CASCADE', ondelete='CASCADE'))
    name = db.Column(db.String(255), nullable=False)
    purchase_cycle_days = db.Column(db.Integer, default=30) # 購入頻度
    stock_rate = db.Column(db.Float, default=1.0) # 現在の在庫割合
    status = db.Column(db.String(10), nullable=False)

    created_at = db.Column(db.DateTime, nullable=False, default=datetime.now)

Item, ItemUpdateLogのそれぞれにstatusを追加してます。
追加できたらマイグレーションを実行して、DBスキーマを更新します。

# flask db migrate
# flask db upgrade

処理の実装

モデルができたので、更新ログを残していく処理を作ります。

item_update_logsにinsertする処理を実装

まずは、ItemUpdateLogレコードを生成し、dbセッションにaddする部分を作ります。

tbl/app.py

def update_item_logs(item):
    """itemを更新する
    """
    log = models.ItemUpdateLog(
        item_id=item.id,
        item_category_id=item.item_category_id,
        name=item.name,
        purchase_cycle_days=item.purchase_cycle_days,
        stock_rate=item.stock_rate,
        status=item.status
    )

    db.session.add(log)

Itemインスタンスを渡すと、その情報からItemUpdateLogインスタンスを生成し、addする、という処理です。

ここでdb.session.commit()までしていないのは、この処理や、この処理を呼び出す一連のDB更新処理でエラーが発生したときにロールバックするためです。

新規登録/編集/削除時に上記を実行

これまで作ってきた、新規登録/編集/削除の処理に、前項のupdate_item_logsを呼び出す処理を追記していきます。

同時に、DB更新後の処理を、「完了ページに遷移」から「ページトップに遷移」するように、redirectの処理に置き換えています。

tbl/app.py

from flask import Flask, render_template, request, redirect, url_for # redirectを追加
(略)
@app.route('/page/<uuid>/item/create', methods=['POST'])
def create_new_item(uuid):
    """新規アイテムを作成する
    """
    if request.args == "":
        return "Error"

    if request.method != 'POST':
        return # 本来はエラー返したい

    # 作成
    user_page_id = request.form['user_page_id']
    name = request.form['name']

    # 値の更新
    item = models.Item(user_page_id, name=name)

    if request.form['purchase_cycle_days']:
        item.purchase_cycle_days = request.form['purchase_cycle_days']

    if request.form['stock_rate']:
        item.stock_rate = request.form['stock_rate']

    try:
        db.session.add(item)
        db.session.flush()
        update_item_logs(item)

        db.session.commit()

    except Exception as e:
        print(e)
        db.session.rollback()
        raise e
(略)
@app.route('/item/edit/<item_id>/<uuid>', methods=['POST'])
def edit_item(item_id, uuid):
    """アイテムを編集する
    """
    # POSTでない場合は編集ページに遷移
    if request.args == "" or request.method != 'POST':
        return show_edit_item_page(item_id)

    if request.form['user_page_id'] is None: # user_page_idが無い場合は戻す
        return redirect(url_for('index'))

    user_page_id = request.form['user_page_id']

    # UPDATE処理
    item = models.Item.searchById(item_id)

    # item.page_idのuuidと一致するか確認
    if check_is_correct_uuid(item, uuid) is False:
        return "uuid does not match."

    # 値の更新
    item.name = request.form['name']
    item.purchase_cycle_days = request.form['purchase_cycle_days']
    item.stock_rate = request.form['stock_rate']


    try:
        db.session.add(item)
        update_item_logs(item)

        db.session.commit()

        return redirect(url_for('show_user_page', uuid=uuid))

    except Exception as e:
        print(e)
        db.session.rollback()
        raise e
(略)
@app.route('/item/delete/<item_id>/<uuid>')
def delete_item(item_id, uuid):
    """アイテムを削除する
    item_idだけで削除できるとまずいので、uuidも渡す
    """
    item = models.Item.searchById(item_id)

    # item.page_idのuuidと一致するか確認
    if check_is_correct_uuid(item, uuid) is False:
        return "uuid does not match."

    item.status = "deleted"

    try:
        db.session.add(item)
        update_item_logs(item)

        db.session.commit()

        return redirect(url_for('show_user_page', uuid=uuid))

    except Exception as e:
        print(e)
        db.session.rollback()
        raise e

各処理の中で、db.session.add()のあとにupdate_item_logs(item)が呼び出され、その後にdb.session.commit()として変更内容を確定されているのが確認できると思います。

また、

    except Exception as e:
        print(e)
        db.session.rollback()
        raise e

ではDB更新中に例外が発生したら更新処理を巻き戻す(=ロールバックする)ようにしています。

新規登録時、flush()でautoincrementされたidを取得する

新規登録(create_item)において、

    try:
        db.session.add(item)
        db.session.flush()
        update_item_logs(item)

という処理について説明します。

ItemUpdateLogには、item_idも含めたItemレコードの情報を格納していますが、新規アイテムでItemレコードを生成したタイミングでは、まだそのitem_idを知る術がありません。
レコードをDBに挿入して初めて、idが振られるからです。

上記コードでいうと、db.session.add(item)しているitemのitem_idはまだ分からないわけです。そのitemをそのままupdate_item_logs(item)してしまうと、item_idが無いためエラーになります。

そこで、新規作成の場合はレコードをaddした後に、db.session.flush()を実行することで、変数itemの中身をDBの中身と同期します。

addすると、挿入されたレコードにはitem_idが振られるので、その情報と同期したことでitem変数にはitem_idを持ったItemインスタンスが格納されている、という状態になり、update_item_logs()を実行してもOKになります。

編集時、データに変更がなければ編集しない処理

次に編集時の処理を見直します。
具体的には、現行処理だと編集内容が編集前と全く同じであっても更新処理が走ってしまい、更新ログが残ってしまう問題が発生します。

更新処理の前に、内容を更新前と比較し、差分がなければ更新処理自体を行わないように書き換えます。

tbl/app.py

(略)
@app.route('/item/edit/<item_id>/<uuid>', methods=['POST'])
def edit_item(item_id, uuid):
    """アイテムを編集する
    """
    # POSTでない場合は編集ページに遷移
    if request.args == "" or request.method != 'POST':
        return show_edit_item_page(item_id)

    if request.form['user_page_id'] is None: # user_page_idが無い場合は戻す
        return redirect(url_for('index'))

    user_page_id = request.form['user_page_id']

    # UPDATE処理
    item = models.Item.searchById(item_id)

    # item.page_idのuuidと一致するか確認
    if check_is_correct_uuid(item, uuid) is False:
        return "uuid does not match."

    # 変更内容差分比較
    bef_values = [
        item.name, item.purchase_cycle_days, item.stock_rate
    ]
    aft_values = [
        request.form['name'], request.form['purchase_cycle_days'], request.form['stock_rate']
    ]
    if has_same_values(bef_values, aft_values):
        return redirect(url_for('show_user_page', uuid=uuid))

    # 値の更新
    item.name = request.form['name']
    item.purchase_cycle_days = request.form['purchase_cycle_days']
    item.stock_rate = request.form['stock_rate']

    try:
        db.session.add(item)
(略)
def has_same_values(before_values, after_values):
    """アイテム情報一式が同じか
    """
    if len(before_values) != len(after_values):
        print('Length of the lists are different')
        return False

    # after_valuesはすべて文字列になっているため、before_valuesをキャストする
    before_values = [str(before_value) for before_value in before_values]

    return before_values == after_values

変更前/変更後の値を比較し、一緒ならば一覧ページにリダイレクトする、という処理です。

ここで比較の処理に詰まりました。

ユーザー入力として送られてくるrequest.form[‘xxx’]の中身は、文字列として送られてきます。なので、変更前の値がDBに格納されているint型の値だったとしても、文字列として比較しないといけません。
has_same_values()の中では、before_values(変更前の値)はstr()で文字列にキャストをした上で、after_values(変更後の値)と比較をしています。

当然、桁数の多い小数などになってくるとこの手法だと限界が出てくるわけですが、今回はそこまで厳密には管理しないので、この手法を採用しています。

動作確認

アイテムを追加

↓

アイテムを編集

値を変更してSubmit

↓

値はそのままでSubmit

(見た目は「変更してSubmit」と変わりません)

アイテムを削除

↓

MySQLのレコードを確認

$ docker-compose exec db mysql -u root -p


mysql> select * from item_update_logs;
...
| 27 |      35 |                1 | シリアル                    |                   5 |          1 | 2020-12-08 06:07:45 | active  |
| 28 |      35 |                1 | シリアル                    |                   5 |        0.5 | 2020-12-08 06:08:05 | active  |
| 29 |      35 |                1 | シリアル                    |                   5 |        0.5 | 2020-12-08 06:08:29 | deleted |
+----+---------+------------------+-----------------------------+---------------------+------------+---------------------+---------+

アイテム追加ログは残っている
一度目(値変更した場合)のアイテム変更ログは残っている
二度目(値変更していない場合)のアイテム変更ログは残っていない
削除ログは残っている
ことが確認できました。

これで更新ログの実装ができました。

まとめ

これで、

更新ログの実装
ページ遷移をしないでレコード追加・更新ができるようにする

が完了しました。
ページ遷移しないで済むようになったので、だいぶ快適です。

ただ、このままだとレコードの変更が行われたのか全く分かりません。完了ページが果たしていた役割を別のもので補う必要があります。
次回は、「直前でどんな変更が入ったのかなどを通知する機能」(Flashといいます)を入れます。

参考

プロダクト開発, 開発autoincrement, Flask, flush, redirect, 消耗品買い物リスト

[pandas]DataFrameのNaNをサクッとNoneに置き換える方法2つ

[Python]setで集合演算(和集合/差集合/積集合)

コメント一覧

まだ、コメントがありません