Chapter 04

共通部分の作成

alivelimb
alivelimb
2022.05.26に更新

前置きが長くなりましたが、本章から YaEC の実装に入っていきましょう。本章では全画面の共通部分である、SessionManager, MockDB, Model, Application を実装していきます。

YaECシステム全体像

SessionManger は Streamlit のセッションとデータをやり取りします。YaEC では見通しの良いコードを目指し、st.session_stateのデータ読取・書込の責務を SessionManager で担うようにします。

MockDB はバックエンドの代わりにローカル(Streamlit が動作している環境)で用意する DB です。本章では DB に接続するインターフェスの定義までを行います。

Application はページの表示切り替えを行います。既に紹介した通り、Streamlit の本来の機能では複数ページを扱えません。そこで複数ページあるように見せる責務を Application で担うようにします。また、サイドバーに選択ボックスを用意することでページの切り替えを可能にします。

SessionManager

st.session_stateとのデータのやり取りは以下の用に行うことが出来ます。

# 読み取り
st.session_state.key
st.session_state["key"]

# 書き込み
st.session_state.key = value
st.session_state["key"] = value

しかし、この形式でやり取りをすると TypeHint がなく、IDE の恩恵を受けることが出来ません。そのため、session_state とデータのやり取りをする際は、タイポが無いように注意して実装する必要がありバグの原因になります。

そこで YaEC では session_state の Wrapper として SessionManager を用意します。

# ※一部抜粋
# ※UserはModelで定義
class StreamlitSessionManager:
    def __init__(self) -> None:
        self._session_state = st.session_state

    def get_user(self) -> User:
        return self._session_state["user"]

    def set_user(self, user: User) -> None:
        self._session_state["user"] = user

Enum を活用する

SessionManager を用意することで、SessionManager を利用する側からはタイポの心配が減り、IDE の恩恵も受けることが出来るようになりました。これでも良いのですが、SessionManager 内の利用ではタイポの心配があるのと、変数が増えてきた時にsession_stateで何を管理しているか分かりにくくなります。

そこで Enum 型でセッションキーを定義しておきます。

# ※一部抜粋
class SessionKey(Enum):
    USER = auto()


class StreamlitSessionManager:
    def get_user(self) -> User:
        return self._session_state[SessionKey.USER.name]

    def set_user(self, user: User) -> None:
        self._session_state[SessionKey.USER.name] = user

文字列から Enum を利用することでタイポの心配が減り、session_stateでの管理対象が増えてきても、SessionKeyを見れば何を管理している分かるようになりました。

MockDB

モックのためのローカル DB にはdatasetTinyDBを利用します。YaEC ではこれらを直接使うのではなく Wrapper として MockDB と MockSessionDB を用意します。

MockDB

MockDB は dataset の Wrapper です。DB の初期化、DB への接続が主な責務になります。

class MockDB:
    def __init__(self, dbpath: Path) -> None:
        s_dbpath = str(dbpath)
        self._dbname = f"sqlite:///{s_dbpath}"
        self._init_mock_db() # 今後の章で実装

    @contextmanager
    def connect(self) -> Generator[dataset.Database, None, None]:
        db = dataset.connect(self._dbname)
        db.begin()
        try:
            yield db
            db.commit()
        except Exception as e:
            db.rollback()
            raise e

dbpath = Path("sample.db")
mock_db = MockDB(dbpath)

# With句でテーブルの定義&データの追加
with mock_db.connect() as db:
    table: dataset.Table = db["samples"]
    print(table.find_one(id="001"))

MockSessionDB

MockSessionDB は TinyDB の Wrapper です。MockDB と同じく DB の初期化、DB への接続が主な責務になります。

class MockSessionDB:
    def __init__(self, dbpath: Path) -> None:
        self._db = TinyDB(dbpath)

    @contextmanager
    def connect(self) -> Generator[TinyDB, None, None]:
        try:
            yield self._db
        except Exception as e:
            raise e

dbpath = Path("sample.json")
mock_session_db = MockSessionDB(dbpath)

with mock_session_db.connect() as db:
    query = Query()
    print(db.search(query.session_id == session_id))

Model

Model ではデータクラスを定義します。データクラスはデータの保持だけでなく、jsonへのエンコード/デコードを行う必要があります。そこでto_dict, from_dictメソッドを持つ BaseDataModel を定義しておきます。

class BaseDataModel(Protocol):
    def to_dict(self) -> dict[str, str]:
        pass

    @classmethod
    def from_dict(cls, data: dict[str, str]) -> BaseDataModel:
        pass

実際にデータクラスを用意する場合は Python の標準パッケージであるdataclassesを用いて定義します。また、BaseDataModel を満たすようにto_dict, from_dictを実装します。

@dataclass
class Sample:
    no: int
    name: str

    def to_dict(self) -> dict:
        ...

    @classmethod
    def from_dict(cls, data: dict) -> BaseDataModel:
        ...お降り

なぜ dataclasses を用いるのか

Python でデータクラスを作りたい場合にはいくつかの選択肢があります。よく使われるのはdataclasses, namedtuple, pydanticが挙げられます。

namedtuple はその名の通り、名前付きのタプルです。タプルは変更不可(イミュータブル)のオブジェクトなので、データの保持によく用いられます。しかし以下の 2 点から今回は dataclasses の方が適していると判断しました。

  1. dataclasses でもfrozen=Trueにすることで、イミュータブルに出来る
  2. dataclasses の判定はis_dataclassで行えるが、namedtuple の判定は自分で実装する必要がある(参考)

次に pydantic です。pydantic は dataclasses よりも非常に高機能ですが、標準パッケージではありません。pydantic を使っても良いと思いますが、今回は dataclasses で事足りたのでインストールが不要なこちらを選択しました。

Application

Application では複数ページを管理する MultiPageApp を用意します。その前に 1 つ 1 つのページの元となる BasePage を定義しておきます。

class BasePage:
    def __init__(self, page_id: PageId, title: str, ssm: StreamlitSessionManager) -> None:
        self.page_id = page_id.name
        self.title = title
        self.ssm = ssm

    def render(self) -> None:
        pass

ページクラスを作成する時は BasePage を継承し、renderを実装します。そして以下に示す MultiPageApp がページクラスを切り替えることで、複数ページアプリのように見せることができます。

class MultiPageApp:
    def __init__(self, ssm: StreamlitSessionManager, pages: list[BasePage], nav_label: str = "ページ一覧") -> None:
        self.pages = {page.page_id: page for page in pages}
        self.ssm = ssm
        self.nav_label = nav_label

    def render(self) -> None:
        # ページ選択ボックスを追加
        page_id = st.sidebar.selectbox(
            self.nav_label, # 選択ボックスのラベル
            list(self.pages.keys()), # ページ一覧
            format_func=lambda page_id: self.pages[page_id].title,
            key=SessionKey.PAGE_ID.name, # (B)
        )

        # ページ描画
        try:
            self.pages[page_id].render() # (A)
        except YaoyaError as e:
            st.error(e)

selectboxではformat_funcを指定することで、ページ ID をそのまま表示させるのではなく、ページタイトルを表示できます。また、ページが切り替わる流れは以下の通りです。

  1. ユーザがページを選択する
  2. page_idが選択されたページの ID になる
  3. コード中(A)でrenderが実行されるページクラスが切り替わる

PageId

SessionManager と同様に Enum を使って PageId を定義しておきましょう。

class PageId(Enum):
    PAGE_ID = auto()

ページを追加したい時は、以下の手順で行います。

  1. PageId の属性を増やす
  2. BasePage を継承したページクラスを作る
  3. MultiPageApp のpagesにページ ID とページクラスのペアを追加する

まとめ

最後に本章で作成した部品、また前章までで紹介したテクニックを組み合わせておきます。

main.py

まずはエントリポイントとなる main.py です。アプリケーションを立ち上げるにはstreamlit run main.pyを実行します。

if not st.session_state.get("is_started", False): # 初期化しているかの確認
    ssm = init_session() # session_stateの初期化
    pages = init_pages(ssm) # ページの初期化
    app = init_app(ssm, pages) # アプリケーションの初期化
    st.session_state["is_started"] = True
    st.session_state["app"] = app
    st.set_page_config(page_title="八百屋さんEC", layout="wide") # Streamlitのページ設定

app = st.session_state.get("app", None)
if app is not None:
    app.render()

init_app.py

次に init_app.py です。main.py で出てきた初期化関数(init_session, init_pages, init_app)を定義します。

# session_stateの初期化
def init_session() -> StreamlitSessionManager:
    mockdir = Path(TemporaryDirectory().name) # (A)
    mockdir.mkdir(exist_ok=True)
    mockdb = MockDB(mockdir.joinpath("mock.db"))
    session_db = MockSessionDB(mockdir.joinpath("session.json"))
    ssm = StreamlitSessionManager()
    return ssm

# ページの初期化
def init_pages(ssm: StreamlitSessionManager) -> list[BasePage]:
    pages = [
        # ページクラスを追加
    ]
    return pages

# アプリケーションの初期化
def init_app(ssm: StreamlitSessionManager, pages: list[BasePage]) -> MultiPageApp:
    app = MultiPageApp(ssm, pages)
    return app

コード中の(A)について補足しておきます。MockDB はあくまで一時的なものであるため、永続化する必要はありません。そこでtempfile.TemporaryDirectoryを使うことで、一時的なディレクトリを作成しています。tempfileに関しては記事にしているので、適宜参照してください。