はじめに
streamlit の初リリース（2018年）から6年が経過し、現在では社内で業務アプリとして活用されるシーンも多く見られるようになりました。
今回は、データベースと連携した streamlit によるフルスタック開発に興味のある方を対象に、画像処理データ管理アプリを作成した内容を共有させていただきたいと思います。
対象読者 / 想定シーン
- 業務のデータ管理アプリをstreamlitで開発したいエンジニア
- streamlitの基本は把握しており、ORMやDBはこれから触っていこうと思っているエンジニア
- pythonだけでフルスタックに開発したいエンジニア
成果物
share.streamlit.ioにデプロイしています。
※デプロイ先の仕様上、停止している場合があります。
停止している場合は、Yes, get this app back upのボタンを押下してください🙇
https://imaima-image-process-manager.streamlit.app/
- UI構成
↓リポジトリ
注意
本アプリにはエラーハンドリング処理が完全に実装されておりません。
業務でご使用される場合は、必ずエラーハンドリング処理を適切に実装してください。
設計 / 採用技術
- フロントエンド: Streamlit
- ORM: SQLAlchemy
- データベース: SQLite
- デザイン: Material Design Icons
- 画像処理: OpenCV
アーキテクチャ
クライアントはstreamlitのUIを経由して、services dir. or database dir.の関数を呼び出す形でタスクを処理する構成としました。
ディレクトリ構成
React(Next.js)の開発経験を元にstreamlitにおいても関数(services, utils)を分離する構成を選択しました。気になる点がありましたら、コメントでお知らせください
streamlit-image-process-manager/
├─ .streamlit/ # streamlit標準config
│ └─ config.toml
├─ data/ # 画像保存先
│ ├─ processed/
│ └─ raw/
├─ database/ # データベース設定
│ ├─ cruds/
│ │ ├─ image_data.py
│ │ └─ processed_image_data.py
│ ├─ database.py
│ └─ models.py
├─ schemas/ # スキーマディレクトリ
│ └─ schemas.py
├─ services/ # フロントエンドから呼び出す関数を保存
│ ├─ delete_image_data.py
│ ├─ dummy_heavy_image_processing.py
│ └─ initialize_setting.py
├─ utils/ # 汎用関数
│ └─ format_datetime_column.py
├─ pages/
│ ├─ 1_🚀_processed_check_page.py
│ ├─ 2_📑_bonus_comment_page.py
│ └─ 3_⚠️_README.py
├─ .env # 環境設定
├─ app.py # メインアプリケーション
└─ data.db # データベース(SQLite3)
コード解説
app.py
- streamlitのメインのソースコードが書かれています。
- インタラクティブで柔軟なデータグリッドを簡単に追加するために、streamlit-aggridを採用しました。
2025/1/5更新 :
AG Gridを商用で使用する場合は、AG Grid Enterpriseのライセンスが必要となります。詳しくは、以下のリンクをお確かめください。
https://www.ag-grid.com/license-pricing/
- AgGridの設定は、GridOptionsBuilder を使用することで効率的に実装することができました。以下のコードが該当する箇所となります。
- また、ユーザーが選択したデータ情報をst.session_stateを使用することで、簡単に状態管理できる状態となっています。
app.py
image_data_list = get_all_image_data(db=db)
# データフレームの作成
if len(image_data_list) > 0:
df_raw_img = pd.DataFrame(image_data_list)
df_raw_img["is_processed"] = df_raw_img["is_processed"].apply(
lambda x: "済" if x else "未"
)
df_raw_img = format_datetime_column(df_raw_img, "uploaded_at")
df_raw_img = format_datetime_column(df_raw_img, "updated_at")
# AgGridのsetting
grid_builder = GridOptionsBuilder.from_dataframe(
df_raw_img, editable=False, filter=True, resizable=True, sortable=False
)
grid_builder.configure_selection(selection_mode="multiple", use_checkbox=True)
grid_builder.configure_side_bar(filters_panel=True, columns_panel=False)
grid_builder.configure_default_column(
enablePivot=True, enableValue=True, enableRowGroup=True
)
grid_builder.configure_grid_options(rowHeight=50)
grid_options = grid_builder.build()
grid_options["columnDefs"][0]["checkboxSelection"] = True
response = AgGrid(
df_raw_img,
gridOptions=grid_options,
update_mode=GridUpdateMode.MODEL_CHANGED,
theme="alpine",
)
# 選択したデータ情報を取得
selected_df = pd.DataFrame(response["selected_rows"])
if not selected_df.empty:
selected_id_list = selected_df["id"].tolist()
st.session_state.selected_id_list = selected_id_list
services/
- フロントエンドからの呼び出しに応じる関数を格納します
services/
├─ delete_image_data.py
├─ dummy_heavy_image_processing.py # ダミーとして重たい画像処理を行います
└─ initialize_setting.py # .envを読み込み、st.session_stateにパスを設定
- UIから呼び出す重たい画像処理 (今回はdummyとしてtime.sleep(5)で代替) としています。
services/dummy_heavy_image_processing.py
def dummy_heavy_image_processing(image: np.ndarray) -> np.ndarray:
"""dummyの重たい画像処理をソーベルフィルターとタイマーで実装"""
time.sleep(5) #5sec間、停止
image = cv2.medianBlur(image, 5)
sobel_x = cv2.Sobel(image, cv2.CV_32F, 1, 0, ksize=3)
sobel_y = cv2.Sobel(image, cv2.CV_32F, 0, 1, ksize=3)
sobel_x = cv2.convertScaleAbs(sobel_x)
sobel_y = cv2.convertScaleAbs(sobel_y)
sobel_combined = cv2.addWeighted(sobel_x, 0.5, sobel_y, 0.5, 0)
return sobel_combined
schemas/
- データベースへの書き込み時に使用するschemaをdataclassで実装しています。
- dataclassの勉強は、以下の記事が大変参考になります。
database/
database/
├─ cruds/
│ ├─ image_data.py # 画像処理前のCRUD操作
│ └─ processed_image_data.py # 画像処理後のCRUD操作
├─ database.py # データベースの接続設定
└─ models.py # テーブルの定義
- 画像処理前のテーブルと、画像処理後のテーブルを以下のように定義しています。
file_name, file_path などは本来uniqueであると思いますので、本番環境では注意してください。
database/models.py
class ImageData(Base):
__tablename__ = "image_data"
id = Column(Integer, primary_key=True, index=True)
file_name = Column(String, index=True)
file_path = Column(String, index=True)
file_extension = Column(String, index=True)
file_size = Column(Float)
description = Column(String)
is_processed = Column(Boolean, default=False)
uploaded_at = Column(
DateTime, default=pd.Timestamp.now(tz="UTC").tz_convert("Asia/Tokyo").floor("s")
)
updated_at = Column(
DateTime, default=pd.Timestamp.now(tz="UTC").tz_convert("Asia/Tokyo").floor("s")
)
def to_dict(self):
return {
"id": self.id,
"file_name": self.file_name,
"file_path": self.file_path,
"file_extension": self.file_extension,
"file_size": self.file_size,
"uploaded_at": self.uploaded_at,
"updated_at": self.updated_at,
"is_processed": self.is_processed,
"description": self.description,
}
class ProcessedImageData(Base):
__tablename__ = "processed_image_data"
id = Column(Integer, primary_key=True, index=True)
file_name = Column(String, index=True)
file_path = Column(String, index=True)
processed_at = Column(
DateTime, default=pd.Timestamp.now(tz="UTC").tz_convert("Asia/Tokyo").floor("s")
)
def to_dict(self):
return {
"id": self.id,
"file_name": self.file_name,
"file_path": self.file_path,
"processed_at": self.processed_at,
}
追加予定(・・・の可能性あり)
- AI画像認識実行機能
最後に
- 今回の構成を参考にし、マイクロサービスアーキテクチャの設計思想に則り、バックエンドをAPI(DRF, Flask, FastAPIなど)で実装してみるのも良いかもしれません。
参考
ORMの操作が慣れていないかたは、公式のチュートリアルを実践してみることをお勧めいたします。英語ですが大変わかりやすいです。
ICONはstreamlitでもGoogloeFontsが使えるので、是非採用してください。