一. 前言：按了重新整理，資料卻沒變？ #
拍拍君今天遇到一個很像靈異事件的問題。 SQL 查得到新資料，Streamlit 表格卻一直顯示舊值。 重跑 App、切換頁面，都沒有用。
兇手不是資料庫，也不是瀏覽器。 是我們只記得「連線可以共用」，忘記「查詢結果也可能被快取」。
今天不做另一個 CRUD 後台。 我們用很小的資料集拆清楚三條邊界：
st.connection管連線資源。conn.query管讀取結果與快取。- 短生命週期 Session 管交易。
UI 可以簡單，資料新鮮度不能靠猜。
二. 安裝與可重現的專案 #
uv init pypy-sql-connection
cd pypy-sql-connection
uv add streamlit sqlalchemy pandas
uv add --dev pytest
mkdir -p .streamlit
本篇使用 SQLAlchemy 2.x 寫法。 SQLite 已包含在 Python 裡，不必另外啟動伺服器。 需要 PostgreSQL 時，還要裝對應 driver，不能只改 URL。
專案只有四個 Python 檔案：
pypy-sql-connection/
├── app.py
├── seed.py
├── queries.py
├── test_queries.py
└── .streamlit/secrets.toml
這是查詢與交易契約的範例，不是多人產品的完整架構。
套件版本請交給 uv.lock 固定；別把「今天最新」當作可重現版本。
三. Named Connection：名字要對上 Secrets #
建立 .streamlit/secrets.toml：
[connections.pypy]
url = "sqlite:///pypy.db"
程式使用 st.connection("pypy", type="sql")。
pypy 對應 [connections.pypy]，不是資料表名稱。
自訂名稱要指定 type="sql"；不要期待名稱自動猜中類型。
SQLite 的相對路徑以啟動程式的工作目錄為準。 因此以下指令都在專案根目錄執行。
真實資料庫的 URL 可能包含密碼。 請忽略實際 Secrets 檔，只提交不含憑證的範本：
.streamlit/secrets.toml
*.db
*.db-journal
部署端另外設定 Secrets；不要印出完整連線 URL 除錯。 如果同時用 kwargs 與 Secrets 設定同一欄位，要明確知道覆蓋來源。 這一點不是格式問題，是「你究竟連到哪個資料庫」的問題。
完整部署流程可接著看 Streamlit 部署與 Secrets。
四. 先做一份不會重跑就清空的資料 #
建立 seed.py，只在初始化時執行：
from sqlalchemy import create_engine, text
engine = create_engine("sqlite:///pypy.db")
with engine.begin() as db:
db.execute(text("""
CREATE TABLE IF NOT EXISTS tasks (
id INTEGER PRIMARY KEY,
owner TEXT NOT NULL,
status TEXT NOT NULL
)
"""))
db.execute(text("""
CREATE TABLE IF NOT EXISTS audit (
task_id INTEGER NOT NULL,
new_status TEXT NOT NULL
)
"""))
count = db.scalar(text("SELECT COUNT(*) FROM tasks"))
if count == 0:
db.execute(
text("INSERT INTO tasks VALUES (:id, :owner, :status)"),
[
{"id": 1, "owner": "拍拍君", "status": "review"},
{"id": 2, "owner": "拍拍醬", "status": "done"},
],
)
uv run python seed.py
不要把重設資料庫塞進 App 的頂層。
Streamlit rerun 是正常互動，不是重新安裝應用程式。
正式 schema 變更仍應使用 migration，不靠 CREATE TABLE IF NOT EXISTS 補洞。
五. 分清楚兩種 TTL #
下面兩個 ttl 看起來相同，責任卻不同：
import streamlit as st
conn = st.connection("pypy", type="sql", ttl=3600)
rows = conn.query("SELECT id, status FROM tasks", ttl=30)
| 設定位置 | 控制的對象 | 不代表什麼 |
|---|---|---|
st.connection(..., ttl=3600) |
快取的連線物件 | 結果每小時才更新 |
conn.query(..., ttl=30) |
查詢結果 | 每 30 秒主動輪詢 |
TTL 到期不是背景計時器。 需要下一次讀取，才有機會重新執行 SQL。 如果畫面沒有 rerun，它不會自己長出新資料。
這裡刻意給查詢明確的 30 秒期限。 預設沒有期限的讀取，很容易把「快」變成「永遠看舊資料」。 官方 API 的快取語意可對照 SQLConnection.query。
六. 參數化：值不要拼到 SQL 裡 #
建立 queries.py，先放共用查詢：
from sqlalchemy import text
TASK_SQL = """
SELECT id, owner, status
FROM tasks
WHERE owner = :owner
ORDER BY id
LIMIT :row_limit
"""
def read_tasks(conn, owner: str, *, fresh: bool = False):
return conn.query(
TASK_SQL,
params={"owner": owner, "row_limit": 20},
ttl=0 if fresh else 30,
show_spinner=False,
)
owner 是資料值，交給 bind parameter。
這樣輸入裡有引號，也不會變成另一段 SQL。
不同查詢參數也應產生各自的快取結果，稍後會測這件事。
參數化不會替你處理 table name、排序欄位或 SQL 關鍵字。
動態排序應先映射固定白名單，不能把使用者文字當識別字。
LIMIT 也不能省：快取不是無限制下載資料的理由。
本篇用 SQLite 的 SQLAlchemy named binds。 換資料庫時，仍須確認 driver 與 SQL dialect 的支援範圍。
七. 最小 App：讓新鮮度成為明確選項 #
建立 app.py：
import streamlit as st
from queries import read_tasks
st.title("拍拍君 SQL 連線觀察室")
conn = st.connection("pypy", type="sql")
owner = st.selectbox("示範資料擁有者", ["拍拍君", "拍拍醬"])
fresh = st.checkbox("即時讀取，不沿用查詢快取", value=False)
st.button("重新讀取")
rows = read_tasks(conn, owner, fresh=fresh)
st.caption("即時模式" if fresh else "查詢快取最長 30 秒")
st.dataframe(rows, hide_index=True)
uv run streamlit run app.py
按鈕會觸發 rerun，但不承諾清掉結果快取。 在 30 秒內按下去，仍可能看到同一份資料。 勾選即時模式，才是本範例的明確 fresh-read 契約。
ttl=0 會讓這條讀取不沿用舊結果，但不是清空所有別人的快取。
退回 30 秒模式時，原本的快取仍可能在有效期限內。
如果你的產品要求寫入後立刻一致，請直接走 fresh read，別只呼叫 rerun。
八. 寫入與交易：不要用 query 假裝更新 #
conn.query 的介面是讀取便利工具，不是交易入口。
需要寫入時，使用短 Session，並在同一個交易內完成相關操作。
在 queries.py 加入下面函式：
def mark_done(conn, task_id: int) -> None:
with conn.session as session:
with session.begin():
result = session.execute(
text("UPDATE tasks SET status = :s WHERE id = :id"),
{"s": "done", "id": task_id},
)
if result.rowcount != 1:
raise LookupError("找不到唯一任務")
session.execute(
text("INSERT INTO audit VALUES (:id, :s)"),
{"id": task_id, "s": "done"},
)
正常離開 session.begin() 時提交；例外則 rollback。
任務更新與 audit 寫入一起成功，才算完成。
詳細語意可看 SQLAlchemy 交易文件。
這是示範交易原子性，不是可重複執行的事件處理器。
同一任務重複呼叫會多一筆 audit；正式系統要定義冪等鍵或狀態轉移規則。
也不要把 Session 放進 session_state，等使用者按鈕等十分鐘。
九. Commit 不等於快取失效 #
資料庫提交後，舊的查詢快取不會因為資料表變更自動收到通知。 因此讀取政策必須獨立設計：
- 報表允許 30 秒延遲：保留短 TTL，畫面清楚註記。
- 寫入後確認結果：用即時讀取，不沿用舊結果。
- 多處資料一起更新：在服務層設計一致的失效策略。
conn.reset() 主要處理連線重建，不是跨使用者的資料失效協定。
不要讓「重連成功」偷偷變成「所有列表一定新鮮」的產品承諾。
st.cache_data.clear() 是粗粒度操作。
共用 App 使用它時，可能影響其他查詢與使用者，不能當作每次提交的預設動作。
資料規模大時，應考慮自有可控快取層，而不是不斷製造不同 key。
十. 可執行的行為測試 #
建立 test_queries.py。
以下用暫存 SQLite 與固定資料，避免碰到開發資料庫。
from uuid import uuid4
import pytest
import streamlit as st
from sqlalchemy import text
from queries import mark_done, read_tasks
@pytest.fixture
def conn(tmp_path):
db_path = tmp_path / "test.db"
conn = st.connection(
f"test_{uuid4().hex}", type="sql", url=f"sqlite:///{db_path}"
)
with conn.engine.begin() as db:
db.execute(text("CREATE TABLE tasks (id INTEGER PRIMARY KEY, owner TEXT, status TEXT)"))
db.execute(text("CREATE TABLE audit (task_id INTEGER, new_status TEXT)"))
db.execute(text("INSERT INTO tasks VALUES (1, '拍拍君', 'review'), (2, '拍拍醬', 'review')"))
yield conn
st.cache_data.clear()
st.cache_resource.clear()
conn.engine.dispose()
def test_params_and_freshness(conn):
first = read_tasks(conn, "拍拍君")
assert first["id"].tolist() == [1]
assert read_tasks(conn, "拍拍醬")["id"].tolist() == [2]
assert read_tasks(conn, "' OR 1=1 --").empty
mark_done(conn, 1)
assert read_tasks(conn, "拍拍君")["status"].tolist() == ["review"]
assert read_tasks(conn, "拍拍君", fresh=True)["status"].tolist() == ["done"]
def test_rollback_if_audit_fails(conn):
with conn.engine.begin() as db:
db.execute(text("DROP TABLE audit"))
from sqlalchemy.exc import SQLAlchemyError
with pytest.raises(SQLAlchemyError):
mark_done(conn, 1)
assert read_tasks(conn, "拍拍君", fresh=True)["status"].tolist() == ["review"]
uv run pytest -q
第一個測試直接重現「commit 了，但快取仍然舊」的情境。 第二個讓 audit 寫入失敗，確認任務更新也一起撤回。 測試裡清除全域快取是隔離測試資料，不是正式 App 的操作建議。 CLI 模式出現沒有 ScriptRunContext 的提示，不等於資料庫測試失敗。
十一. 權限、連線失效與多人使用 #
範例 selectbox 只是在切換示範資料，不是授權。
正式系統的 owner 或 tenant 必須從可信登入身分取得。
篩選條件有參數化，不表示使用者有權看那筆資料。
一般 SQLConnection 與查詢快取可能在同一個 App 行程內共用。 不要把個人權限只藏在某個 Session 變數裡，卻讓查詢快取忽略它。 所有影響可見資料的可信條件，都應進入查詢與快取契約。 更嚴格的隔離需求，可交給資料庫角色、row-level security 或服務 API。
連線偶發失效與權限錯誤，也要分開處理。
前者可能需要 conn.reset() 後重讀；後者需要修正授權，不該盲目重試。
寫入失敗更不能直接無限重送：是否已提交、是否冪等，都要先釐清。
結語：快取很香，但契約要先寫清楚 #
拍拍君最想留下的不是某一條 API，而是這個分工： 連線可以長壽，交易要短，結果的新鮮度要說得清楚。
先測參數不串資料、快取不假裝即時、失敗能 rollback。 這三件事做好，才比較值得把查詢面板交給別人每天使用。