はじめに

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が使えるので、是非採用してください。