NTTドコモ データプラットフォーム部(以下DP部)外山です。
DP部では「『あらゆる業務・現場のニーズに応じられる』柔軟なデータ活用環境」を目指し、社内データ活用プラットフォームPochi*1の開発を進めています。
一つ目のアプリが産声を上げてから2年以上が経過し、様々なアイデアが形となって、名実ともにビジネス部門と共にプロダクトの成長を続けてきました。
そのプロセスにおいて、Streamlitそのものも進化を続けている反面、長年開発してきた私たちの実装上の気づきが眠っています。
特に、アプリ初期開発時点から中規模開発以上に至ったアプリが直面する課題は私たちならではかと思います。
そこで、DP部の松本竜矢さんがこの課題解決に向けて進めていた内容をご紹介させていただきます。 以下は、松本さんに執筆いただいています。
社内データ活用プラットフォームPochiとは
私たちDP部は社内のデータ民主化を目指し、StreamlitとGoogle Cloudで圧倒的に使いやすいデータ活用プラットフォームを開発・推進しています。 このプラットフォームは、24年度は30万時間もの業務効率化を実現し、直近では5,000人以上の社員に利用が拡大しています。
ASCII.jp:NTTドコモ、Streamlit利用の“ポチポチ分析アプリ”開発で社内データ活用を促進 (1/3)
こんにちは、DP部の松本です。私は主にStreamlitを使った社内分析アプリの開発を担当させていただいています。
突然ですが、Streamlitって最高ですよね。アイデアをサッと形にできる手軽さは、一度味わうとやめられません。
私もこれまで数多くのStreamlitアプリを開発してきましたが、プロジェクトの規模が大きくなるにつれて、ある共通の課題に直面してきました。
st.session_stateのキーが文字列なので、typoしても実行するまで気づけない...- どこでどのキーが使われているのか、追いかけるのがだんだん辛くなる...
- UIのコードとロジックがごちゃ混ぜになって、コンポーネントの再利用が難しい...
- 結果として、コードは複雑化し、テストは書きづらくなり、開発効率が頭打ちに...
心当たりのある方も多いのではないでしょうか?
今日は、これらの課題を解決し、中規模以上のStreamlit開発を劇的に効率化する、私が実践している2つの設計パターン、SessionStateProxyとComponentProtocolを紹介したいと思います!
本記事で紹介するコード例は、以下の環境で動作確認しています。
- Python 3.10以降
- Streamlit 1.38以降
よくある実装パターンと、その落とし穴
まずは、よくある分析アプリを例に、課題を具体的に見ていきましょう。
サイドバーで分析条件を入力し、ボタンを押すと処理が実行され、結果がメイン画面に表示される、というシンプルなアプリです。
import time from datetime import date import pandas as pd import streamlit as st # データ読み込み関数 def load_data(source: str, name: str, period: tuple[date, date]) -> pd.DataFrame: time.sleep(1) return pd.DataFrame( { "value": range(10), "log": [f"{source} - {name} - {period[0]} - {i}" for i in range(10)], } ) st.title("分析アプリ") # --- サイドバー --- with st.sidebar: st.header("分析条件") # ウィジェットの値をsession_stateへ格納 st.session_state["analysis_name"] = st.text_input("分析名") st.session_state["analysis_period"] = st.date_input( "分析期間", value=(date.today(), date.today()) ) # データソース選択 source = st.selectbox("データソースを選択", ["データA", "データB"]) if st.button("分析を実行"): # 入力条件を取得する name = st.session_state.analysis_nama # あっ、typoしてる... period = st.session_state.analysis_period # データを読み込み df = load_data(source, name, period) # 結果をsession_stateに格納 st.session_state["processed_data"] = df st.session_state["last_source"] = source # --- メイン画面 --- if "processed_data" in st.session_state: st.success(f"「{st.session_state['last_source']}」の分析が完了しました。") st.dataframe(st.session_state["processed_data"]) else: st.info("分析を実行してください。")
実はこのコード、実行してボタンを押すとAttributeErrorでアプリが止まります。
原因は、ボタンクリック時の処理内でanalysis_nameをanalysis_namaと打ち間違えているからなんですが、こういうtypo、実行するまで気づけないんですよね...。
私もこれまで何度もこのパターンにハマってきました。 そのたびに「文字列キーって便利だけど、開発しづらいな」と感じていたんです。
課題の整理
一見シンプルに見えるコードですが、実は厄介な問題がいくつか潜んでいます。
まず、文字列キーのtypo問題。st.session_state["analysis_nama"]のようなアクセスは、Pyrightのような静的解析ツールでは検知できません。実行して初めて「あっ...」ってなるやつです。デバッグに余計な時間がかかってしまいます。
次に、UIとロジックの密結合問題。分析処理(load_data)を実行するためにsession_stateから値を取得し、UIの表示ロジックもsession_stateの状態で分岐しています。つまり、UI、ロジック、状態がsession_stateを介して密結合しており、部分的な再利用やテストが困難になります。
そして、キー管理の追跡地獄問題。アプリが大きくなり、複数の場所でsession_stateを更新するようになると、「このキーはいつどこで更新されているのか」を追跡するのが非常に困難になります。これがスパゲッティコード化の原因となります。
これらの問題は、最初は小さく見えますが、開発が進むにつれて雪だるま式に膨れ上がっていきます。気づいた時には保守が困難になっているケースが多いのです。
解決策①:SessionStateProxyで状態管理を型安全にする
session_stateの型が分からなくなったり、キー管理が煩雑になったりする問題を解決するために、私が作ってみたのがSessionStateProxyという設計パターンです!
控えめに言って、これを開発してからの開発体験が劇的に変わったので、ぜひ皆さんにも試してほしいです。
これは、st.session_stateをラップして、Pydanticモデルの力を借りて型安全な属性アクセス(ドットアクセス)を可能にするヘルパークラスなんです。
SessionStateProxy本体(クリックで展開)
- クラス定義時にフィールドをスキャン:
__init_subclass__という特殊なメソッドを使い、UserStateのようなクラスが定義された瞬間に、usernameやfavoritesといったフィールド(属性)を自動的に検出します。 - 動的にプロパティを生成: 検出した各フィールドに対して、Pythonの
propertyという機能を使って、裏側でgetter/setterを動的に作成します。- getter:
state.usernameのように値が読み出されたら、st.session_state["user__state__username"]の値を返します。 - setter:
state.username = "John"のように値が書き込まれたら、st.session_state["user__state__username"]にその値を設定します。 - プレフィックスとフィールド名を'__'で連結した一意なキー(例: 'user__state__username')を自動生成することで、他とのキー衝突を防いでいます。
- getter:
st.session_state のことを気にせず、普通のクラスの属性を扱うのと同じ感覚で、型安全な状態管理ができるようになる、というわけです。
from __future__ import annotations from typing import Any, ClassVar, get_type_hints import streamlit as st from pydantic import BaseModel, ConfigDict, Field, PrivateAttr from pydantic.fields import FieldInfo from typing_extensions import Self class SessionStateProxy(BaseModel): """Streamlitのsession_stateをPydanticモデルとして管理するプロキシクラス このクラスを継承することで、型安全なセッション状態管理が可能になります。 各フィールドはStreamlitのsession_stateに自動的に同期され、 プレフィックスによる名前空間の分離により、複数の状態を独立して管理できます。 Attributes: __key_prefix__: session_stateのキーに使用するプレフィックス(ClassVar) Example: >>> class UserState(SessionStateProxy): ... __key_prefix__ = "user" ... username: str = "" ... favorites: list[str] = [] >>> state = UserState() >>> state.username = "John" # st.session_state["user__state__username"] に保存 >>> print(state.username) # "John" >>> >>> # 異なるプレフィックスで独立した状態を管理 >>> state2 = UserState.create(key_prefix="admin") >>> state2.username = "Admin" # st.session_state["admin__state__username"] に保存 """ __key_prefix__: ClassVar[str] = "default" _instance_key_prefix: str = PrivateAttr(default="") # インスタンスごとのkey_prefix model_config = ConfigDict( validate_assignment=True, arbitrary_types_allowed=True, extra="forbid", ) def __init_subclass__(cls, **kwargs: Any): """サブクラス定義時にcallableなデフォルト値をFieldに変換 list, dict, set, tupleなどのミュータブルな型がデフォルト値として 直接指定された場合、各インスタンスで独立したオブジェクトが作成されるよう Field(default_factory=...)に自動変換します。 Args: **kwargs: 親クラスに渡すキーワード引数 """ super().__init_subclass__(**kwargs) # クラスのアノテーションを取得 annotations = cls.__annotations__ if hasattr(cls, "__annotations__") else {} # 各フィールドをチェック for field_name in annotations: if field_name.startswith("_"): continue # デフォルト値を取得 if hasattr(cls, field_name): default_value = getattr(cls, field_name) # callableで、かつ型クラス(list, dict, set など)の場合 if callable(default_value) and default_value in (list, dict, set, tuple): # Field(default_factory=...)に変換 setattr(cls, field_name, Field(default_factory=default_value)) @classmethod def __pydantic_init_subclass__(cls, **kwargs: Any): """Pydantic処理完了後に各フィールドをプロパティに変換""" super().__pydantic_init_subclass__(**kwargs) # 型ヒント情報を取得 annotations = get_type_hints(cls) if hasattr(cls, "__annotations__") else {} # 各フィールドをプロパティに変換 for field_name in annotations: # プライベート属性や特別な属性はスキップ if field_name.startswith("_") or field_name in ("model_config", "__key_prefix__"): continue # プロパティを動的に生成 cls._create_property(field_name) @classmethod def _create_property(cls, field_name: str) -> None: """指定されたフィールド名のプロパティを生成 Args: field_name: プロパティ化するフィールド名 """ def getter(self: SessionStateProxy) -> Any: """session_stateから値を取得""" key = self._get_full_key(field_name) if key not in st.session_state: # デフォルト値を設定 field = self.__class__.model_fields[field_name] st.session_state[key] = self._get_default_value(field) return st.session_state[key] def setter(self: SessionStateProxy, value: Any) -> None: """session_stateに値を設定""" key = self._get_full_key(field_name) st.session_state[key] = value def deleter(self: SessionStateProxy) -> None: """session_stateから値を削除""" key = self._get_full_key(field_name) st.session_state.pop(key, None) # プロパティを設定 prop = property(getter, setter, deleter, doc=f"Property for {field_name}") setattr(cls, field_name, prop) def __init__(self, **data: Any): """SessionStateProxyを初期化 Args: **data: フィールドの初期値 """ # key_prefixを抽出 (createメソッドから渡された場合) key_prefix = data.pop("key_prefix", None) # Pydanticの初期化を先に行う super().__init__(**data) # key_prefixが指定されていれば使用、そうでなければクラス変数を使用 # object.__setattr__を使って直接設定(__setattr__をバイパス) if key_prefix is not None: object.__setattr__(self, "_instance_key_prefix", key_prefix) else: object.__setattr__(self, "_instance_key_prefix", self.__class__.__key_prefix__) self._init_fields() @classmethod def create(cls, key_prefix: str | None = None, **data: Any) -> Self: """カスタムkey_prefixでインスタンスを作成するファクトリーメソッド Note: 型推論を正しく動作させるために、クラスメソッドとして実装。 Args: key_prefix: session_stateのキーに使用するプレフィックス **data: フィールドの初期値 Example: >>> class PageState(SessionStateProxy): ... __key_prefix__ = "page" ... counter: int = 0 >>> state = PageState.create(key_prefix="custom") >>> # stateの型はPageStateとして推論されます """ # key_prefixをdataに含めて__init__に渡す if key_prefix is not None: data["key_prefix"] = key_prefix return cls(**data) def _get_default_value(self, field: FieldInfo) -> Any: """フィールドのデフォルト値を取得する共通メソッド default_factoryがあればそれを使用し、なければdefaultを使用します。 callableなデフォルト値は自動的に呼び出されます。 Args: field: Pydanticのフィールド情報 Returns: デフォルト値 """ # default_factoryがある場合はPydanticのget_default()メソッドを使用 if field.default_factory is not None: return field.get_default(call_default_factory=True) # callablesは初期値として呼び出す default_value = field.default if callable(default_value) and default_value not in ( str, int, float, bool, list, dict, set, tuple, ): return default_value() elif default_value in (list, dict, set): # list, dict, setの型そのものがデフォルト値の場合 return default_value() # type: ignore else: return default_value def _init_fields(self) -> None: """各フィールドをsession_stateに初期登録 session_stateにキーが存在しない場合のみ、デフォルト値で初期化します。 """ for name, field in self.__class__.model_fields.items(): # キーを取得し、存在しない場合はデフォルト値で初期化 key = self._get_full_key(name) if key not in st.session_state: st.session_state[key] = self._get_default_value(field) def _get_full_key(self, name: str) -> str: """プレフィックス付きキーを生成 Args: name: フィールド名 Returns: プレフィックス付きのキー(例: "user__state__username") """ return f"{self._instance_key_prefix}__state__{name}" def __repr__(self) -> str: """デバッグ用のオブジェクト表現""" items = [f"{k}={getattr(self, k)!r}" for k in self.__class__.model_fields] return f"{self.__class__.__name__}({', '.join(items)})"
このクラスを使うと、st.session_state["analysis_name"]というアクセスが、state.analysis_nameのように書けるようになります。
使い方は非常に簡単で、session_state_proxy.pyに上記コードを保存し、以下のように継承して使うだけです。
from datetime import date import streamlit as st from src.session_state_proxy import SessionStateProxy # 状態を管理するクラスを定義 class AnalysisFormState(SessionStateProxy): # 継承する __key_prefix__ = "analysis_form" # キーの衝突を防ぐプレフィックス # 登録・管理したい値を定義 analysis_name: str = "" analysis_period: tuple[date, date] | None = None # インスタンス化するだけで準備完了 state = AnalysisFormState() # アクセスはドットで! state.analysis_name = "ユーザー分析" st.write(state.analysis_name) # "ユーザー分析"
裏では次の処理を自動で行ってくれます。
- session_stateにキーが存在しない場合、Pydanticモデルのデフォルト値で初期化
- 属性アクセス時にsession_stateから値を取得・設定・削除
使ってみて実感した良さ
typoから解放される
state.analisys_nameのようなtypoは、PyrightやPylanceが即座に検出してくれます。
IDEのコード補完も効くため、開発効率が大幅に向上しました。一度慣れると手放せなくなります。
実際に使ってみるとこんな感じです。
# 状態を管理するクラスを定義 class PageState(SessionStateProxy): __key_prefix__ = "analysis_page" # フォームの状態をフィールドとして定義 analysis_name: str = "" analysis_period: Tuple[date, date] | None = (date.today(), date.today()) # インスタンス化するだけで初期化完了 state = PageState() # ドットアクセスで読み書き state.analysis_name = "新しい分析" st.write("分析名:", state.analysis_name) # 型チェックの恩恵 text_date = ("2024-01-01", "2024-12-31") # 文字列のタプル # これはPylanceでエラーにならない(session_stateは型チェックできない) st.session_state["analysis_period"] = text_date # これはPyrightで即座にエラー検出!★型安全性の向上! state.analysis_period = text_date # Error: Type "tuple[str, str]" is incompatible with type "tuple[date, date] | None"
VSCode上では、こんなエラーメッセージが表示されます。
クラス "PageState" の属性 "analysis_period" に割り当てることはできません
型 "tuple[Literal['2024-01-01'], Literal['2024-12-31']]" は型 "tuple[date, date] | None" に割り当てできません
"tuple[Literal['2024-01-01'], Literal['2024-12-31']]" は "tuple[date, date]" に割り当てできません
tuple エントリ 1 の型が正しくありません
"Literal['2024-01-01']" は "date" に割り当てできません
型エラーをコーディング中に検出できることで、開発効率が大きく向上します。
AIとの相性が良い
型定義が明確だと、GitHub Copilotのようなコード生成AIも理解しやすくなります。
実際、導入後はコード生成の精度が明らかに向上し、開発サイクル全体がスピードアップしました。個人的には、これが最も大きなメリットだと感じています。
解決策②:ComponentProtocolで責務を分離する
状態管理は型安全になりましたが、もう一つ解決したい問題がありました。
そう、「UIとロジックが混在してしまう」問題です。
この問題を解決するために考えたのが、ComponentProtocolという設計パターン。
コンポーネントの「インターフェース」を定義するためのProtocolクラスで、UIとロジックをきれいに分離できます。
from abc import ABC, abstractmethod from typing import Generic, Optional, Type, TypeVar T = TypeVar("T") class ComponentProtocol(Generic[T], ABC): """コンポーネントの基底プロトコル""" model: Optional[Type[T]] = None @abstractmethod def render(self, key: str, disabled: bool = False) -> Optional[T]: """コンポーネントを描画し、結果を返す""" pass
このProtocolを継承することで、すべてのコンポーネントが統一されたrenderメソッドを持つことを強制できます。
UIの描画と、そこから得られる値の取得を、責務として明確に分離できます。
使ってみて実感した良さ
再利用しやすくなった
renderを呼ぶだけで使えるため、どんなコンポーネントも同じように扱えます。「このコンポーネントをrenderすれば、描画されて値が返ってくる」という直感的な構造になり、コードの可読性が向上しました。
テストが書けるようになった
Protocolで型が明確になったおかげで、コンポーネント単位でのテストが格段に書きやすくなりました。個人的に、Streamlitアプリのテストが現実的になったのは、このパターンを導入してからです。
コンポーネントテストの実装例(クリックで展開)
from datetime import date from streamlit.testing.v1 import AppTest from src.component import AnalysisConditionComponent, AnalysisCondition def test_analysis_condition_component_render(): """AnalysisConditionComponentが正しく描画され、値を返すかテストする""" def app(): import streamlit as st # テスト対象のコンポーネントをインスタンス化 component = AnalysisConditionComponent() # renderメソッドを呼び出し、結果をsession_stateに保存 result = component.render(key="test") st.session_state["result"] = result # AppTestでアプリを実行 at = AppTest.from_function(app).run() # 初期状態では、デフォルト値でAnalysisConditionが返る initial_result = at.session_state["result"] assert isinstance(initial_result, AnalysisCondition) assert initial_result.name == "" assert initial_result.period[0] == date.today() assert initial_result.source == "データA" # テキスト入力をシミュレート at.text_input(key="test_name").input("テスト分析").run() # 入力後の値が正しく反映されたAnalysisConditionが返ることを確認 updated_result = at.session_state["result"] assert updated_result.name == "テスト分析"
コンポーネント単位でサクッとテストが書けるため、リファクタリングも安心してできるようになります。
活用例:2つのパターンを組み合わせてみる
それでは、SessionStateProxyとComponentProtocolを組み合わせて、最初の課題だらけだったコードをリファクタリングしてみましょう。
1. コンポーネントの返り値を定義
まず、コンポーネントが返すデータの型を定義します。
from datetime import date from pydantic import BaseModel class AnalysisCondition(BaseModel): """分析条件を保持するデータモデル""" name: str period: tuple[date, date] source: str
2. コンポーネントを実装
次に、分析条件を入力するコンポーネントを作ります。
from datetime import date import streamlit as st from pydantic import BaseModel from src.component_protocol import ComponentProtocol class AnalysisCondition(BaseModel): """分析条件を保持するデータモデル""" name: str period: tuple[date, date] source: str class AnalysisConditionComponent(ComponentProtocol[AnalysisCondition]): """分析条件入力コンポーネント""" model = AnalysisCondition def render(self, key: str, disabled: bool = False) -> AnalysisCondition: """コンポーネントを描画し、入力された分析条件を返す""" st.header("分析条件") name = st.text_input("分析名", key=f"{key}_name", disabled=disabled) period = st.date_input( "分析期間", key=f"{key}_period", value=(date.today(), date.today()), disabled=disabled, ) source = st.selectbox( "データソースを選択", ["データA", "データB"], key=f"{key}_source", disabled=disabled, ) return AnalysisCondition(name=name, period=period, source=source)
3. ページ側で組み合わせる
最後に、ページ側で状態管理とコンポーネントを組み合わせます。
import pandas as pd import streamlit as st import time from typing import Optional # 状態管理クラス class PageState(SessionStateProxy): __key_prefix__ = "analysis_page" analysis_result: Optional[AnalysisCondition] = None processed_data: Optional[pd.DataFrame] = None # データ読み込み関数 def load_data(condition: AnalysisCondition) -> pd.DataFrame: time.sleep(1) return pd.DataFrame({ "value": range(10), "log": [f"{condition.source} - {condition.name} - {condition.period[0]} - {i}" for i in range(10)], }) st.title("分析アプリ") # 状態管理を初期化 state = PageState() # サイドバーにコンポーネントを配置 with st.sidebar: condition_component = AnalysisConditionComponent() # コンポーネントは入力値を1つに纏めて返してくれる analysis_input = condition_component.render(key="analysis_condition") # 分析実行ボタン if st.button("分析実行", key="execute_analysis"): if not analysis_input.name: st.error("分析名を入力してください。") else: # 分析条件と結果をステートに保存 state.analysis_result = analysis_input state.processed_data = load_data(analysis_input) # メイン画面に結果を表示 if state.processed_data is not None and state.analysis_result is not None: st.success(f"「{state.analysis_result.source}」の分析が完了しました。") st.dataframe(state.processed_data) else: st.info("分析条件を設定し、「分析実行」ボタンをクリックしてください。")
どう変わったのか
見比べてみると、こんな風に変わりました。
型チェックが効く
state.analysis_resultやcondition.nameなど、すべてのアクセスで型チェックが効くようになりました。typoは実行前に検出できるので安心です。
責務がきれいに分かれた
UIコンポーネント(AnalysisConditionComponent)、状態管理(PageState)、ビジネスロジック(load_data)がきれいに分離されました。それぞれ独立してテストできるのが嬉しいポイント。
実行タイミングが明確
ボタンクリックで分析を実行して、結果をステートに保存する形になったので、「いつ分析が実行されたか」が明確になりました。コンポーネントの変更で勝手に再実行されることなく、ユーザーが意図したタイミングで処理が走ります。
コードが読みやすくなった
「このコンポーネントはAnalysisConditionを返す」「この状態はPageStateで管理されている」と、コードを読むだけで構造が理解できるようになりました。
また、state.analysis_resultに保存された条件を後から参照できるため、メイン画面で「どの条件で分析したか」を表示する際も容易になります。
実際に使用してみると、開発体験が大きく変わります。IDEの補完が効き、AIも理解しやすくなり、テストも書きやすくなります。一度慣れると手放せなくなります。
導入効果とまとめ
この2つの設計パターンを導入してから、Streamlit開発の効率が大きく改善されました。
開発効率の向上
型チェックのおかげで、コーディング中のミスが大幅に減少しました。IDEの補完も効くため、開発スピードが大幅に向上しています。キー名を探し回る時間が不要になったことも、生産性向上に貢献しています。
保守性の向上
UIとロジックが分離され、コンポーネント単位でテストが書けるようになりました。変更に強いコードになり、リファクタリングも安心して行えるようになりました。
AIとの相性が良い
型定義が明確なコードは、GitHub Copilotのようなコード生成AIが理解しやすく、精度が明らかに向上しました。開発サイクル全体がスピードアップし、実装に集中できる時間が増えています。
おわりに
もちろん、ごく小規模な使い捨てアプリであれば、ここまでの設計は過剰かもしれません。
しかし、チームで開発する場合や、長期的にメンテナンスしていく中規模以上のStreamlitアプリであれば、SessionStateProxyとComponentProtocolは開発効率とコード品質を両立させる強力な手段となります。
ただ、いきなりこうした設計を適用するのはハードルが高いと感じる方もいるかもしれません。個人的には、まず動くものを作ってから、後でこうした設計パターンでリファクタリングしていくアプローチを好んでいます。 最初から完璧を目指すよりも、まず動かしてみて、課題が見えてきたら改善していく方が、結果として効率的で、開発体験も向上します。
この記事が、皆さんのStreamlit開発の参考になれば幸いです。
*1:Pochiは社内の開発コードネームです