streamlit で Azure Blob Storage の画像を探しやすくするアプリを作りました

執筆日

2025/02/06

概要

Azure Blob Storageに溜まった画像を確認するときにPortalでポチポチして編集から画像確認したり、Azure Storage Explolerを使ってストリーミング表示したりしていたのですが使い勝手が悪かったのでstreamlitでアプリにしてみました。荒はありますが、あまり労力をかけずにそれなりに使いやすくなった気がします。

出来ること

• 画像の一覧表示
  • 8枚ずつのサムネイル表示と追加表示ボタン
  • 拡大表示・ダウンロード・削除
• コンテナ選択
• フィルタリング
  • prefixフィルタ（多層対応）
  • BLOB名キーワード検索
• 画像データのキャッシュ

前提

• Azure Storage Accountに画像ファイルがある
• 接続文字列を取得できる

依存ライブラリインストール

pip install streamlit azure-storage-blob python-dotenv

スクリプト

アプリ本体のapp.pyとAzure Blob Storage接続用の.envを用意します。（ローカルで使うだけであれば環境変数はベタ打ちでもいいですが……）

app.py

app.py

import os
from io import BytesIO

import streamlit as st
from PIL import Image
from dotenv import load_dotenv

from azure.storage.blob import BlobServiceClient


# --- .env ファイルから接続情報を読み込む ---
load_dotenv("./.env")
AZURE_CONNECTION_STRING = os.environ.get("AZURE_CONNECTION_STRING")

# --- 許可する画像拡張子 ---
ALLOWED_EXTENSIONS = {".jpg", ".jpeg", ".png", ".gif", ".bmp", ".tiff"}

# --- Blob 一覧の取得（コンテナ名をキーに含める） ---
@st.cache_data(ttl=300)
def get_blob_list(container_name: str):
    try:
        client = blob_service_client.get_container_client(container_name)
        blobs = list(client.list_blobs())
        return blobs
    except Exception as e:
        st.error(f"コンテナ '{container_name}' の取得に失敗しました。エラー: {e}")
        return None

# --- Blob から画像データを取得 ---
@st.cache_data(show_spinner=False)
def get_image_data(blob_name: str, container_name: str):
    client = blob_service_client.get_container_client(container_name)
    blob_client = client.get_blob_client(blob_name)
    stream = BytesIO()
    blob_client.download_blob().readinto(stream)
    stream.seek(0)
    image = Image.open(stream)
    return image

# --- Blob のバイトデータ取得（ダウンロード用） ---
@st.cache_data(show_spinner=False)
def get_blob_bytes(blob_name: str, container_name: str):
    client = blob_service_client.get_container_client(container_name)
    blob_client = client.get_blob_client(blob_name)
    data = blob_client.download_blob().readall()
    return data

# --- Azure Blob Storage の接続設定 ---
blob_service_client = BlobServiceClient.from_connection_string(AZURE_CONNECTION_STRING)

# --- サイドバー: 利用可能なコンテナ一覧から選択 ---
containers = [container["name"] for container in blob_service_client.list_containers()]
selected_container = st.sidebar.selectbox("コンテナを選択", containers)

all_blobs = get_blob_list(selected_container)
# エラー時はそれ以降の処理をスキップする
if all_blobs is None:
    st.stop()

# --- 画像ファイルのみ対象にする（拡張子フィルタ） ---
blobs = [blob for blob in all_blobs if os.path.splitext(blob.name)[1].lower() in ALLOWED_EXTENSIONS]

# --- サイドバー: フィルタ ---
st.sidebar.title("フィルタ")
current_prefix = ""
while True:
    current_depth = len(current_prefix.split("/")) if current_prefix else 0
    candidates = set()
    for blob in blobs:
        if current_prefix and not blob.name.startswith(current_prefix + "/"):
            continue
        parts = blob.name.split("/")
        if len(parts) > current_depth + 1:
            candidates.add(parts[current_depth])
    if not candidates:
        break
    sorted_candidates = sorted(candidates)
    selection = st.sidebar.selectbox(
        f"{current_depth+1}層目のプレフィックス",
        ["すべて"] + sorted_candidates,
        key=f"prefix_level_{current_depth+1}"
    )
    if selection == "すべて":
        break
    else:
        current_prefix = f"{current_prefix}/{selection}" if current_prefix else selection

# --- サイドバー: キーワード検索 ---
search_keyword = st.sidebar.text_input("Blob名のキーワード検索", value="")

# --- フィルタリング処理 ---
filtered_blobs = []
for blob in blobs:
    if current_prefix and not blob.name.startswith(current_prefix + "/"):
        continue
    file_name = blob.name.split("/")[-1]
    if search_keyword and search_keyword.lower() not in file_name.lower():
        continue
    filtered_blobs.append(blob)

st.write(f"{len(filtered_blobs)} hits.")

# --- 表示件数の管理（初期8枚、もっと表示ボタンで8枚ずつ追加） ---
if "display_count" not in st.session_state:
    st.session_state.display_count = 8
if st.button("display more"):
    st.session_state.display_count += 8
blobs_to_display = filtered_blobs[:st.session_state.display_count]

# --- サムネイル表示（1行4枚） ---
cols_per_row = 4
rows = [blobs_to_display[i:i+cols_per_row] for i in range(0, len(blobs_to_display), cols_per_row)]
for row in rows:
    cols = st.columns(len(row))
    for col, blob in zip(cols, row):
        image = get_image_data(blob.name, selected_container)
        image_thumb = image.copy()
        image_thumb.thumbnail((200, 200))
        col.image(image_thumb, caption=blob.name.split("/")[-1], use_container_width=True)
        button_cols = col.columns(3)
        if button_cols[0].button("🔍", key=f"enlarge-{blob.name}"):
            st.session_state["selected_image"] = blob.name
        button_cols[1].download_button(
            label="⬇️",
            data=get_blob_bytes(blob.name, selected_container),
            file_name=blob.name.split("/")[-1],
            key=f"download-{blob.name}"
        )
        if button_cols[2].button("🗑️", key=f"delete-{blob.name}"):
            st.session_state["delete_candidate"] = blob.name

# --- 拡大表示 ---
if "selected_image" in st.session_state:
    st.write(f"表示画像BLOB名: {st.session_state['selected_image']}")
    selected_blob_name = st.session_state["selected_image"]
    image = get_image_data(selected_blob_name, selected_container)
    st.image(image, use_container_width=True)

# --- 削除確認の処理 ---
if "delete_candidate" in st.session_state and st.session_state["delete_candidate"]:
    candidate = st.session_state["delete_candidate"]
    st.warning(f"本当に {candidate} を削除しますか？")
    col1, col2 = st.columns(2)
    if col1.button("はい"):
        try:
            client = blob_service_client.get_container_client(selected_container)
            client.delete_blob(candidate)
            st.success(f"{candidate} を削除しました。")
            st.cache_data.clear()
            st.session_state["delete_candidate"] = ""
            st.experimental_rerun()
        except Exception as e:
            st.error(f"削除に失敗しました: {e}")
    if col2.button("いいえ"):
        st.session_state["delete_candidate"] = ""

.env

AZURE_CONNECTION_STRING="<接続文字列>"

起動

ブラウザで自動表示されなかった場合はコンソールに表示されるURLをブラウザに入力してください。

$ streamlit run app.py

>>  You can now view your Streamlit app in your browser.

>>  Local URL: http://localhost:8501
>>  Network URL: http://<ip>:8501

表示例

こんな感じです。サイドバーでコンテナ選択、プレフィックス選択（階層構造を動的に追加）やキーワードによるフィルタリング。各画像に対して、拡大表示・ダウンロード・削除ができます。動作のサクサク感はストレージのプランや種類によります。（安いプランだと画像表示までにちょっと時間がかかります……）

参考

今回のアプリはChatGPT o3-miniを使って作成しました。（streamlitのドキュメント全く読んでないです、ごめんなさい）アプリ概要と要件を提示して大枠を作った後、サイドバー周りや各ボタンの挙動や表示のブラッシュアップ指示をして数往復でこれくらいのものが作れて楽しいです。コーディング自体はGPT-4oでも全然可能なレベルですが、回答速度と指示の理解力に差があるのと思考過程が表示されるのが良かったです。