Textual + DuckDB 實戰：終端機資料 Dashboard 小工具

Python 學習 - 本文屬於一個選集。

§ 72: 本文

一. 前言：有些 Dashboard 留在終端機更順手
#

很多資料小工具最後都會走向 Dashboard。你可能有一包每天更新的 CSV，也可能有一批 Parquet log，還可能只是想快速看幾個統計數字：

今天有多少筆資料？
哪個類別最多？
最近幾筆異常長什麼樣子？
查詢結果能不能用鍵盤捲動？
不開瀏覽器也能不能操作？如果使用者本來就在 terminal 工作，每次開瀏覽器其實有點打斷節奏。這時候可以用 Textual + DuckDB 做一個終端機資料 Dashboard。 Textual 負責 TUI：畫面 layout、鍵盤操作、表格、狀態列、按鈕。 DuckDB 負責查詢：直接讀 CSV/Parquet、跑 SQL、做 group by、取 preview。這篇不做 CRUD。也不做完整 Web App。我們要做的是一個唯讀分析工具：選資料檔、執行幾個固定查詢、顯示 summary cards 和表格。如果你想補背景，可以先看： Python Textual 實戰、 Python DuckDB 實戰、 Streamlit + DuckDB Dashboard。今天的角度不同。 Streamlit 版適合給非工程同事在瀏覽器操作。 Textual 版適合工程師自己在 terminal 裡快速巡資料。拍拍君覺得兩者都值得放進工具箱。

二. 建立專案
#

先建立專案：

uv init textual-duckdb-dashboard
cd textual-duckdb-dashboard
uv add textual duckdb pandas pyarrow

不用 uv 的話也可以：

python -m venv .venv
source .venv/bin/activate
pip install textual duckdb pandas pyarrow

專案結構先保持簡單：

textual-duckdb-dashboard/
├── app.py
└── data/
    └── orders.csv

建立 data/orders.csv：

order_id,order_date,customer,region,category,quantity,unit_price
1001,2026-06-01,拍拍醬,North,book,2,12.5
1002,2026-06-01,chatPTT,South,keyboard,1,79.0
1003,2026-06-02,小拍,East,notebook,5,4.2
1004,2026-06-02,拍拍君,West,mouse,2,25.0
1005,2026-06-03,拍拍醬,North,monitor,1,199.0
1006,2026-06-03,chatPTT,East,book,3,12.5

資料量很小沒關係。我們先把架構做乾淨。之後你換成幾十萬列 CSV 或 Parquet， DuckDB 仍然可以直接查。

三. 先把 DuckDB 查詢包起來
#

在 TUI 程式裡，拍拍君不建議到處散落 SQL 字串。先把資料存取集中在幾個函式裡。建立 app.py，先寫資料層：

from __future__ import annotations
from dataclasses import dataclass
from pathlib import Path
import duckdb
import pandas as pd
DATA_FILE = Path("data/orders.csv")
@dataclass(slots=True)
class DashboardStats:
    orders: int
    revenue: float
    average_order_value: float
def table_expr_for(path: Path) -> str:
    suffix = path.suffix.lower()
    if suffix == ".csv":
        return "read_csv_auto(?)"
    if suffix == ".parquet":
        return "read_parquet(?)"
    raise ValueError("只支援 CSV 與 Parquet")

table_expr_for() 看起來很小，但它很重要。我們只允許 CSV 和 Parquet。不要把任意字串直接塞進 SQL。接著寫一個共用查詢 helper：

def query_df(path: Path, sql: str, params: list[object] | None = None) -> pd.DataFrame:
    table_expr = table_expr_for(path)
    full_sql = sql.format(table=table_expr)
    with duckdb.connect() as con:
        return con.execute(full_sql, [str(path), *(params or [])]).df()

這裡有一個小技巧： {table} 只由我們自己的 table_expr_for() 產生，檔案路徑仍然用 ? 參數傳入。所以呼叫端可以寫：

query_df(
    DATA_FILE,
    "SELECT * FROM {table} ORDER BY order_date DESC LIMIT 20",
)

最後執行的概念會像：

SELECT * FROM read_csv_auto(?) ORDER BY order_date DESC LIMIT 20

這比自己拼路徑安全。

四. Summary 查詢
#

現在寫三個查詢：

總體指標。
region summary。
最近訂單。

def load_stats(path: Path) -> DashboardStats:
    df = query_df(
        path,
        """
        SELECT
            count(*) AS orders,
            sum(quantity * unit_price) AS revenue,
            avg(quantity * unit_price) AS average_order_value
        FROM {table}
        """,
    )
    row = df.iloc[0]
    return DashboardStats(
        orders=int(row["orders"]),
        revenue=float(row["revenue"] or 0),
        average_order_value=float(row["average_order_value"] or 0),
    )
def load_region_summary(path: Path) -> pd.DataFrame:
    return query_df(
        path,
        """
        SELECT
            region,
            count(*) AS orders,
            round(sum(quantity * unit_price), 2) AS revenue
        FROM {table}
        GROUP BY region
        ORDER BY revenue DESC
        """,
    )
def load_recent_orders(path: Path, limit: int = 20) -> pd.DataFrame:
    return query_df(
        path,
        """
        SELECT
            order_id,
            order_date,
            customer,
            region,
            category,
            quantity,
            unit_price,
            round(quantity * unit_price, 2) AS total
        FROM {table}
        ORDER BY order_date DESC, order_id DESC
        LIMIT ?
        """,
        [limit],
    )

這三個函式都回傳普通 Python 物件或 DataFrame。好處是：你可以不用啟動 Textual，直接在 Python REPL 或測試裡檢查查詢結果。例如：

if __name__ == "__main__":
    print(load_stats(DATA_FILE))
    print(load_region_summary(DATA_FILE))

先讓資料層能跑，再處理 UI。 TUI bug 已經夠有趣了，不要讓 SQL bug 一起加入。

五. 建立 Textual App 骨架
#

接著加入 Textual。先 import 需要的元件：

from textual.app import App, ComposeResult
from textual.containers import Container, Horizontal, Vertical
from textual.widgets import DataTable, Footer, Header, Static

然後寫一個最小 App：

class OrderDashboard(App):
    CSS = """
    Screen {
        layout: vertical;
    }
    #stats {
        height: 5;
        padding: 1 2;
        background: $surface;
        border: round $primary;
    }
    #tables {
        height: 1fr;
    }
    .panel {
        border: round $accent;
        padding: 1;
    }
    """
    BINDINGS = [
        ("r", "refresh", "重新整理"),
        ("q", "quit", "離開"),
    ]
    def compose(self) -> ComposeResult:
        yield Header(show_clock=True)
        yield Static("載入中...", id="stats")
        with Horizontal(id="tables"):
            with Vertical(classes="panel"):
                yield Static("Region Summary")
                yield DataTable(id="region-table")
            with Vertical(classes="panel"):
                yield Static("Recent Orders")
                yield DataTable(id="orders-table")
        yield Footer()

這個 layout 分成三段：

Header：標題與時間。
Stats：總體指標。
Tables：左邊 region summary，右邊最近訂單。先不用追求漂亮。讓資料正確出來比較重要。

六. 把 DataFrame 塞進 DataTable
#

Textual 的 DataTable 需要明確加入欄位與列。我們寫一個 helper：

def fill_table(table: DataTable, df: pd.DataFrame) -> None:
    table.clear(columns=True)
    for column in df.columns:
        table.add_column(str(column))
    for row in df.itertuples(index=False):
        table.add_row(*(format_cell(value) for value in row))
def format_cell(value: object) -> str:
    if pd.isna(value):
        return "-"
    if isinstance(value, float):
        return f"{value:,.2f}"
    return str(value)

format_cell() 可以先寫得保守。 Dashboard 最怕一開始就塞太多格式化規則。等你知道使用者真的需要什麼，再加顏色、單位、百分比。接著在 App 裡載入資料：

class OrderDashboard(App):
    # CSS / BINDINGS / compose 先省略
    def on_mount(self) -> None:
        self.refresh_dashboard()
    def refresh_dashboard(self) -> None:
        stats = load_stats(DATA_FILE)
        region_df = load_region_summary(DATA_FILE)
        orders_df = load_recent_orders(DATA_FILE)
        stats_widget = self.query_one("#stats", Static)
        stats_widget.update(
            "  ".join(
                [
                    f"Orders: {stats.orders:,}",
                    f"Revenue: {stats.revenue:,.2f}",
                    f"Avg: {stats.average_order_value:,.2f}",
                ]
            )
        )
        fill_table(self.query_one("#region-table", DataTable), region_df)
        fill_table(self.query_one("#orders-table", DataTable), orders_df)
    def action_refresh(self) -> None:
        self.refresh_dashboard()
        self.notify("資料已重新整理")

最後加上啟動點：

if __name__ == "__main__":
    OrderDashboard().run()

啟動：

uv run python app.py

現在你應該可以在 terminal 裡看到兩張表。用滑鼠或方向鍵捲動。按 r 重新整理。按 q 離開。

七. 讓工具更可靠：錯誤、參數與固定查詢
#

實務上，資料檔可能不存在，欄位可能改名，CSV 也可能壞掉。不要讓 App 直接炸成 traceback。最小做法是加一個訊息區，再把 refresh 包起來：

def refresh_dashboard(self) -> None:
    message = self.query_one("#message", Static)
    try:
        stats = load_stats(self.data_file)
        region_df = load_region_summary(self.data_file)
        orders_df = load_recent_orders(self.data_file)
    except Exception as exc:
        message.update(f"讀取失敗：{exc}")
        self.notify("讀取資料失敗", severity="error")
        return
    message.update(f"資料來源：{self.data_file}")
    self.update_stats(stats)
    fill_table(self.query_one("#region-table", DataTable), region_df)
    fill_table(self.query_one("#orders-table", DataTable), orders_df)

資料檔也不要寫死。可以用 argparse 讓使用者指定 CSV 或 Parquet：

import argparse
def parse_args() -> argparse.Namespace:
    parser = argparse.ArgumentParser()
    parser.add_argument("path", nargs="?", default="data/orders.csv")
    return parser.parse_args()
if __name__ == "__main__":
    args = parse_args()
    OrderDashboard(Path(args.path)).run()

之後就能這樣跑：

uv run python app.py data/orders.csv
uv run python app.py data/orders.parquet

最後提醒一件事：第一版工具先用固定 query，比任意 SQL 輸入框更穩。任意 SQL 很自由，但也代表：結果欄位不固定、錯誤訊息更複雜、使用者可能跑出巨大結果。固定 query 則可以針對表格欄位、排序、空資料提示和格式化慢慢打磨。你可以先提供幾個 action：

r：重新整理。
1：region summary。
2：category summary。
3：recent orders。
/：簡單搜尋。等需求真的長大，再把 SQL editor 做成進階模式。

八. 效能與適用情境
#

Textual 本身不是資料庫。不要把幾百萬列全部塞進 DataTable。比較穩的做法：

Summary 交給 DuckDB 聚合。
Raw table 只顯示最近 50 到 500 筆。
大表格用分頁或搜尋。
重新整理時只重跑需要的 query。
長查詢放進 worker，避免 UI 卡住。這個組合適合工程師自己用、團隊內部用、SSH 到 server 看結果，也適合幫 CLI 工具加一個 inspect mode。如果使用者多半不是工程師，或你需要登入、權限、多人共用、手機瀏覽器存取，那就回到 Streamlit、FastAPI + 前端，或正式 BI 工具。拍拍君的判斷很簡單：資料在本機或 server，使用者也在 terminal，Textual + DuckDB 很舒服。使用者想用瀏覽器點來點去，Streamlit 比較親切。

九. 小結：把資料檢查變成一個可用工具
#

今天我們用 Textual + DuckDB 做了一個終端機資料 Dashboard。核心思路是：

DuckDB 直接查 CSV/Parquet。
SQL 集中在資料層。
Textual 負責互動畫面。
DataTable 顯示 summary 和最近資料。
固定 query 比任意 SQL 更適合第一版工具。這不是要取代 Streamlit。它是另一種工作流：當你本來就在 terminal，想快速檢查資料，又不想只看一串 print()， Textual + DuckDB 就很剛好。先從唯讀 dashboard 開始。等它真的幫你省時間，再慢慢加搜尋、分頁、背景查詢和匯出。