Streamlit Form 提交生命週期：草稿、已套用狀態與失敗保留

uv init --python 3.12
uv add 'streamlit==1.66.1'
uv add --dev 'pytest==9.2.0' 'pandas==3.1.0'

from dataclasses import dataclass
import re

CATALOG = ("拍拍咖啡", "拍拍餅乾", "拍拍筆記本")
DATA_VERSION = "demo-v1"

@dataclass(frozen=True)
class Query:
    needle: str = ""
    limit: int = 2

@dataclass(frozen=True)
class Snapshot:
    query: Query
    rows: tuple[str, ...]
    data_version: str

def parse_query(needle: str, limit_text: str) -> Query:
    needle = needle.strip()
    if len(needle) > 20:
        raise ValueError("關鍵字最多 20 字")
    if not re.fullmatch(r"[1-3]", limit_text):
        raise ValueError("筆數必須是 1、2 或 3")
    return Query(needle, int(limit_text))

def lookup(query: Query) -> tuple[str, ...]:
    if query.needle == "!timeout":
        raise TimeoutError("示範查詢逾時")
    return tuple(x for x in CATALOG if query.needle in x)[:query.limit]

def execute(needle: str, limit_text: str) -> Snapshot:
    query = parse_query(needle, limit_text)
    rows = lookup(query)
    return Snapshot(query, rows, DATA_VERSION)

from dataclasses import asdict
import streamlit as st
from query import Query, execute

st.set_page_config(page_title="拍拍君提交實驗室")
ss = st.session_state
for key, value in {
    "ui_needle": "", "ui_limit": "2", "snapshot": None,
    "error": "", "attempts": 0, "successes": 0, "runs": 0,
}.items():
    if key not in ss:
        ss[key] = value

def submit():
    ss["attempts"] += 1
    try:
        candidate = execute(ss["ui_needle"], ss["ui_limit"])
    except (ValueError, TimeoutError) as exc:
        ss["error"] = str(exc)
        return
    ss["snapshot"] = candidate
    ss["successes"] += 1
    ss["error"] = ""

def reset(clear_result=False):
    if clear_result:
        ss["snapshot"] = None
        ss["attempts"] = 0
        ss["successes"] = 0
    old = ss["snapshot"]
    query = old.query if old is not None else Query()
    ss["ui_needle"] = query.needle
    ss["ui_limit"] = str(query.limit)
    ss["error"] = ""

ss["runs"] += 1
st.title("拍拍君提交實驗室")
st.button("無關重跑", key="rerun_only")
st.button("還原輸入", key="restore", on_click=reset)
st.button("全部重設", key="reset_all", on_click=reset, args=(True,))

with st.form("search", clear_on_submit=False, enter_to_submit=True):
    st.text_input("關鍵字", key="ui_needle")
    st.text_input("筆數（1～3）", key="ui_limit")
    st.form_submit_button("套用查詢", key="apply", on_click=submit)

if ss["error"]:
    st.error(ss["error"])
snapshot = ss["snapshot"]
if snapshot is None:
    st.info("尚未套用查詢")
else:
    st.subheader("上次成功套用")
    st.write(f"條件：{snapshot.query.needle!r}；筆數：{snapshot.query.limit}")
    st.write(f"資料版本：{snapshot.data_version}")
    st.table({"名稱": list(snapshot.rows)})
    if not snapshot.rows:
        st.info("查詢成功，但沒有符合資料")

st.caption("以下只代表伺服器已知值，不是瀏覽器即時草稿")
st.json({
    "runs": ss["runs"], "attempts": ss["attempts"],
    "successes": ss["successes"],
    "server_needle": ss["ui_needle"], "server_limit": ss["ui_limit"],
    "applied": asdict(snapshot) if snapshot is not None else None,
})

from dataclasses import FrozenInstanceError
import pandas  # 先載入 st.table 使用的原生相依，避免計入 AppTest 單輪逾時
import pytest
from streamlit.testing.v1 import AppTest
from query import execute

@pytest.mark.parametrize("limit", ["0", "4", "02", "２"])
def test_bad_limit(limit):
    with pytest.raises(ValueError):
        execute("拍拍", limit)

def start():
    at = AppTest.from_file("app.py", default_timeout=15).run()
    assert not at.exception
    return at

def apply(at, needle, limit="2"):
    at.text_input(key="ui_needle").set_value(needle)
    at.text_input(key="ui_limit").set_value(limit)
    at.button(key="apply").click().run()
    assert not at.exception

def test_initial_and_unrelated_rerun():
    at = start()
    at.button(key="rerun_only").click().run()
    assert at.session_state["attempts"] == 0
    assert at.session_state["snapshot"] is None
    assert at.session_state["runs"] == 2

def test_fresh_values_and_frozen_snapshot():
    at = start()
    apply(at, "咖啡", "1")
    snap = at.session_state["snapshot"]
    assert (snap.query.needle, snap.query.limit) == ("咖啡", 1)
    assert snap.rows == ("拍拍咖啡",)
    with pytest.raises(FrozenInstanceError):
        snap.query.limit = 3

@pytest.mark.parametrize("needle,limit", [("餅乾", "0"), ("!timeout", "2")])
def test_failed_attempt_keeps_success(needle, limit):
    at = start()
    apply(at, "咖啡", "1")
    before = at.session_state["snapshot"]
    apply(at, needle, limit)
    assert at.session_state["snapshot"] == before
    assert at.session_state["ui_needle"] == needle
    assert (at.session_state["attempts"], at.session_state["successes"]) == (2, 1)
    assert len(at.error) == 1
    assert at.table[0].value["名稱"].tolist() == ["拍拍咖啡"]
    at.button(key="rerun_only").click().run()
    assert at.session_state["snapshot"] == before
    assert at.session_state["attempts"] == 2

def test_restore_then_reset():
    at = start()
    apply(at, "咖啡", "1")
    apply(at, "!timeout")
    at.button(key="restore").click().run()
    assert at.text_input(key="ui_needle").value == "咖啡"
    assert at.text_input(key="ui_limit").value == "1"
    assert not at.error
    assert at.session_state["attempts"] == 2
    at.button(key="reset_all").click().run()
    assert not at.exception
    assert at.session_state["snapshot"] is None
    assert at.text_input(key="ui_needle").value == ""
    assert at.text_input(key="ui_limit").value == "2"
    assert at.session_state["attempts"] == at.session_state["successes"] == 0

def test_zero_rows_is_success_and_clears_error():
    at = start()
    apply(at, "!timeout")
    assert at.session_state["snapshot"] is None
    apply(at, "不存在")
    assert not at.error
    assert at.session_state["snapshot"].rows == ()
    assert at.session_state["successes"] == 1

uv run pytest -q
uv run streamlit run app.py