こんにちは、本日もpythonでWebサービスを作ってみます。
簡単なWebサービスをPythonのみで作ってみた１では、ストラテジーパターンを適用して、画面を作成しましたが、今回は内部状態を持たせ、状態に応じてそれぞれのメニューを表示する、といったことをジコマンで勝手にやっていきます。
サマリ
さてまずは今回やったことのまとめです。
- これまでの画面に1画面追加する
- 追加した画面にボタンを４つ（初期、新設、取得、削除）を生成する。
- 初期、新設、取得、削除のいずれかを押下したとき、内部状態を変化させる。
- それぞれのボタンの応じた画面を表示する。
画面イメージ
完成した画面は以下の通りです。
初期状態の画面
「新規」ボタン押下後の画面
「取得」ボタン押下後の画面
「削除」ボタン押下後の画面
もちろんそれぞれのボタン押下後の画面での処理も実施できます。例えば「取得」ボタン押下後の画面で「シーケンス番号」指定して「照会」ボタンを押下すると以下のように表示されます。
今回、このようなボタンによる画面の切替をステートパターンを用いて実装しました。
実装コード
さてここからコードの説明に入ります。アプリは簡単なWebサービスをPythonのみで作ってみた２までのものを前提に、追加開発しています。
※もし全量が欲しい場合は連絡いただけば、githubに招待いたします。
・view_strategy.py
前回の繰り返しになりますが、新しい画面を追加するには、
- 「view_strategy.py」に１つクラスを新設
- 「menu.properties」にその定義を追加
で対応可能であり、既存のコードに一切手を入れずに機能追加ができます。
from abc import ABCMeta, abstractmethod
import json
import requests
import streamlit as st
import pandas as pd
from states import *
class IView(metaclass=ABCMeta):
_url: str = "http://localhost:8000/api/v2/users/"
@abstractmethod
def show_screen(self) -> None:
pass
・
・
<中略>
・
・
@st.cache_resource
class ViewCustom(IView):
__state: IState = None
__state_manage = StateManage()
disabled_msg = "未許可遷移のため、遷移できなかったでござるよ"
def show_screen(self) -> None:
if not self.__state:
self.__state = InitialState.get_instance()
col0, col1, col2, col3 = st.columns(4)
if col0.button("初期"):
if self.__state_manage.is_transitionable(self.__state.get_classname(), States.InitialState):
self.__state = InitialState.get_instance()
else:
st.write(self.disabled_msg)
if col1.button("新規"):
if self.__state_manage.is_transitionable(self.__state.get_classname(), States.NewAddState):
self.__state = NewAddState.get_instance()
else:
st.write(self.disabled_msg)
if col2.button("取得"):
if self.__state_manage.is_transitionable(self.__state.get_classname(), States.GetDataState):
self.__state = GetDataState.get_instance()
else:
st.write(self.disabled_msg)
if col3.button("削除"):
if self.__state_manage.is_transitionable(self.__state.get_classname(), States.DeleteState):
self.__state = DeleteState.get_instance()
else:
st.write(self.disabled_msg)
self.__state.show()
{
"home": "ViewHome",
"inquire": "ViewUser",
"inquireAll": "ViewAllUsers",
"register": "ViewRegister",
"update": "ViewUpdate",
"deleteAll": "ViewDeleteAll",
"custom": "ViewCustom"
}さて少しソースについてみていきます。
今回「ViewCustom」クラスを新設してます。このクラスは他のクラスと違って、デコレータ「@st.cache_resource」を追加しています。これは通常Streamlitはボタンをform_submit_buttonを押下するとインスタンスの再読み込みが走ります。その影響で内部の状態が常に初期化されてしまうのですが、「@st.cache_resource」をつけることで、リソースをキャッシュできるようになります。
ステートパターンにおける「Context」クラスが「ViewCustom」に相当します。
「__state」が内部状態を表すインスタンス変数です。外から隠蔽するためprivateにしてます。
「__state_manage」が状態遷移を管理するクラスのインスタンス変数です。これは後述します。
「show_screen」メソッドで、初期、新規、、、のボタンを配置と押下時の挙動を定義し、それ以降の各状態の画面ロジックは内部状態「__state」のshowメソッドに委譲しています。
states.py
次に状態に関する機能はstates.pyに入れてます。コードは以下の通りです。
from abc import abstractmethod
from enum import Enum
import json
from typing import Any
import requests
import streamlit as st
# 状態の基底クラス
class IState:
_instance = None
kbn: str = '基底'
_url: str = "http://localhost:8000/api/v2/users/"
def __init__(self):
raise NotImplementedError('Cannot Generate Instance By Constructor')
@classmethod
def get_instance(cls):
if not cls._instance:
cls._instance = cls.__new__(cls)
return cls._instance
@classmethod
def get_classname(cls):
return States[cls.__name__]
@abstractmethod
def show(self) -> None:
pass
# Conscreteクラス 初期状態
class InitialState(IState):
kbn: str = "初期状態ですぞ〜"
def show(self) -> None:
st.header(self.kbn)
return
# Conscreteクラス 新規登録状態
class NewAddState(IState):
kbn: str = "データをぶっこむですぞ"
def show(self) -> None:
st.header(self.kbn)
with st.form(key="custom"):
seq_nbr: str = st.text_input("シーケンス番号")
first_name: str = st.text_input("First Name")
last_name: str = st.text_input("Last Name")
gender: str = st.radio(
label="性別を選択せよ",
options=("female", "male"),
index=0,
horizontal=True,
)
roles: str = st.radio(label="ロールを選択せよ", options=("admin", "user"))
payload = {
"sequence_nbr": seq_nbr,
"first_name": first_name,
"last_name": last_name,
"gender": gender,
"roles": roles,
}
submit_bottun: bool = st.form_submit_button(label="登録")
if submit_bottun:
res = requests.post(url=self._url, json=json.dumps(payload))
if res.status_code == 200:
st.success("登録完了")
st.markdown(res.text)
# Conscreteクラス 取得状態
class GetDataState(IState):
kbn: str = "データをゲットするですぞ"
def show(self) -> None:
st.header(self.kbn)
with st.form(key="custom"):
seq_nbr: str = st.text_input("シーケンス番号", key="seq")
submit_bottun = st.form_submit_button(label="照会")
if submit_bottun:
res = requests.get(self._url + seq_nbr)
decorded_text = res.content.decode("utf-8")
list_data = json.loads(decorded_text)
if not isinstance(list_data, list):
list_data = [list_data]
st.table(list_data)
# Conscreteクラス 削除状態
class DeleteState(IState):
kbn: str = "データを削除しまするぞ"
def show(self) -> None:
st.header(self.kbn)
with st.form(key="custom"):
submit_bottun: bool = st.form_submit_button(label="削除")
if submit_bottun:
res = requests.delete(self._url)
if res.status_code == 200:
st.success("削除完了")
st.markdown(res.text)
# 状態の列挙クラス
class States(Enum):
InitialState = "InitialState"
NewAddState = "NewAddState"
GetDataState = "GetDataState"
DeleteState = "DeleteState"
# 状態の管理クラス
class StateManage:
# 可能な遷移を予め定義
transition_enabled: dict = {}
transition_enabled[States.InitialState] = {States.NewAddState, States.GetDataState, States.DeleteState}
transition_enabled[States.NewAddState] = {States.GetDataState, States.DeleteState}
transition_enabled[States.GetDataState] = {States.InitialState, States.NewAddState, States.DeleteState}
transition_enabled[States.DeleteState] = {States.GetDataState}
# 可能な遷移か否か判定
def is_transitionable(self, from_state: States, to_state: States) -> bool:
is_enabled = False
if from_state in self.transition_enabled:
enabled_to_state_set: dict = self.transition_enabled[from_state]
if to_state in enabled_to_state_set:
is_enabled = True
return is_enabled
詳細はコードの通りなので、ポイントをいくつか述べておきます。
まず「IState」クラスを基底クラスにして、これを継承して、ConcreteStateクラスを実装しています。また状態はいくつもインスタンスを生成する必要もないので、シングルトンにしてます。
ただし、マルチスレッドの場合はシングルトンの保証が完全ではないので、注意してください。スレッドAがインスタンス生成中にスレッドBがインスタンス生成を開始すると、２つインスタンスが作られてしまいます。
とりうる状態はなるべく仕掛けで縛るようにするためEnumクラスで状態をまとめています。さらに可能な状態遷移をdict型(他の言語ならmap型)でまとめてしまってます。これによって、今後状態を新たに新設したとしても、既存への影響をなしにできます。
遷移可能かどうかは「is_transitionable」での判定を提供することで、Context側(VeiwCustomクラス側)で実施するようにさせています。
最後に
ステートパターンはストラテジーパターン同様、リファクタリングのツールになったりします。身につけてみるとかなりすっきりしたコードになったります。
最後まで読んでいただきありがとうございます。
質問等はコメント欄かお問合せにてよろしくおねがいいたします。