Prefab のリアクティブ UI コンポーネントと静的 HTML エクスポートを活用した Python フォーカスのインタラクティブダッシュボード設計方法
Prefab を使用して Python のみでインタラクティブなダッシュボードを構築し、React パワーの UI を記述せずに静的 HTML としてエクスポートする手法が紹介されています。
キーポイント
Python 単一環境での UI デザイン
Prefab のコンポーネントベースインターフェースを用いて、React や JavaScript を書かずにチャート、テーブル、フィルターなどの複雑な UI を Python コードだけで構築できます。
リアルタイムデータ連携とリアクティブ性
生成されたパイプライン監視データをリアルタイムの UI コントロールに接続し、状態管理（State）やイベント駆動型のアクションを可能にするリアクティブなダッシュボードを実現します。
Google Colab での完全ワークフロー
インストールからコード記述、実行まで Google Colab 環境内で完結し、最終成果物を静的 HTML ファイルとしてエクスポートしてブラウザで直接プレビュー可能です。
データ生成と集計ロジックの統合
ランダムシードを用いて30日間の地域別パイプライン実行データを擬似的に生成し、日次・地域別に成功率やROIを計算する集計関数を定義しています。
動的な状態管理とアクション設計
地域切り替え時に `SetState` を用いて行データやKPIを即座に更新し、ユーザーフィードバックとして `ShowToast` によるトースト通知を実装しています。
多様な可視化データの準備
散布図やレーダーチャート用の整形済みデータセットを構築し、地域ごとのパフォーマンス（成功率、コスト、レイテンシ）を比較分析可能な形式で用意しています。
Prefab コンポーネントの主要インポート
Prefab のコンポーネント、アクション、チャート、制御フロー、およびリアクティブ状態処理のための主要なインポートを追加します。
影響分析・編集コメントを表示
影響分析
このアプローチは、データ分析と可視化の壁を取り払い、Python コードを書くだけでプロフェッショナルな UI を提供できるため、開発者の生産性を大幅に向上させる可能性があります。特に、フロントエンド専門職を必要としない小規模チームや研究環境において、迅速なプロトタイピングと実装を可能にする重要なツールとなります。
編集コメント
フロントエンド開発の負担を減らし、データロジックに集中できる環境を提供する点で実用性が高い記事です。ただし、大規模な複雑なシステムへの適用には、既存の React エコシステムとの整合性を確認する必要があります。
このチュートリアルでは、Python で完全にインタラクティブなダッシュボードを作成する方法を示す Prefab アプリケーションの構築を行います。Prefab のコンポーネントベースの Python インターフェースを使用して、リアクティブな状態、チャート、テーブル、フィルター、フォーム、タブ、アラート、メトリクス、クライアントサイドアクションを備えた洗練された運用ダッシュボードを設計します。リアルなパイプライン監視データを生成し、ライブ UI コントロールに接続し、最終的なアプリを静的 HTML ダッシュボードとしてエクスポートして、Google Colab 内で直接プレビューできるようにします。このワークフローを通じて、Prefab を使用すれば、フロントエンドコードを手動で記述することなく、Python のデータロジックからモダンな React パワーのユーザーインターフェースへ移行できる方法を学びます。
Colab での Prefab インストール
コードをコピーしました別のブラウザを使用してください
import os
import sys
import base64
import subprocess
from pathlib import Path
from IPython.display import HTML, display, FileLink
PREFAB_VERSION = "0.20.2"
APP_PATH = Path("/content/prefab_advanced_tutorial_app.py")
HTML_PATH = Path("/content/prefab_advanced_dashboard.html")
subprocess.check_call([
sys.executable,
"-m",
"pip",
"install",
"-q",
f"prefab-ui=={PREFAB_VERSION}",
])
APP_CODE = ""
Colab 環境を設定するために、必要な Python ユーティリティをインポートし、Prefab のバージョン、アプリのパス、および HTML エクスポートのパスを定義します。チュートリアルがバージョン関連の問題なく一貫して実行されるように、固定された prefab-ui パッケージをインストールします。また、Prefab アプリケーションを段階的に構築するために使用する空の APP_CODE 文字列も初期化します。
コードをコピーしました
ブラウザを変更する
APP_CODE += r'''
合成オペレーションデータの生成
コードをコピーしました
ブラウザを変更する
import random
from collections import Counter, defaultdict
from datetime import date, timedelta
from prefab_ui.actions import AppendState, OpenLink, PopState, SetState, ShowToast, ToggleState
from prefab_ui.app import PrefabApp
from prefab_ui.components import (
Alert, AlertDescription, AlertTitle, Badge, Button, Card, CardContent,
CardDescription, CardFooter, CardHeader, CardTitle, Code, Column,
DataTable, DataTableColumn, Form, Grid, H2, Input, Markdown, Mermaid,
Metric, Muted, Progress, Ring, Row, Slider, Small, Switch, Tab, Tabs,
Text
)
from prefab_ui.components.charts import (
BarChart, ChartSeries, LineChart, PieChart, RadarChart, ScatterChart,
Sparkline
)
from prefab_ui.components.control_flow import Else, ForEach, If
from prefab_ui.rx import EVENT, STATE
random.seed(42)
TODAY = date.today()
DATES = [TODAY - timedelta(days=29 - i) for i in range(30)]
REGIONS = ["All", "APAC", "EMEA", "NA", "LATAM"]
PIPELINES = [
"Customer 360 ETL",
"Invoice OCR",
"LLM Triage",
"Risk Scoring",
"Forecast Sync",
"Warehouse Load",
]
OWNERS = ["Data Platform", "AI Apps", "Revenue Ops", "Risk Engineering"]
STATES = ["Completed", "Completed", "Completed", "Completed", "Late", "Failed"]
PRIORITIES = ["P0", "P1", "P2", "P3"]
runs = []
daily_region_rows = []
for d in DATES:
for region in REGIONS[1:]:
region_bias = {
"APAC": 0.96,
"EMEA": 0.94,
"NA": 0.97,
"LATAM": 0.91,
}[region]
volume = random.randint(32, 78)
failures = 0
late = 0
total_cost = 0.0
total_latency = 0.0
total_revenue = 0.0
for i in range(volume):
pipeline = random.choice(PIPELINES)
owner = random.choice(OWNERS)
state = random.choices(
STATES,
weights=[
region_bias * 10,
6,
4,
3,
1.2,
max(0.2, (1 - region_bias) * 16),
],
k=1,
)[0]
duration = max(
12,
int(
random.gauss(95, 35)
+ (20 if state == "Late" else 0)
+ (45 if state == "Failed" else 0)
),
)
cost = round(max(0.09, random.lognormvariate(-1.15, 0.55) + duration / 1800), 2)
revenue = round(random.uniform(1.2, 8.5) * (1.3 if state == "Completed" else 0.6), 2)
priority = random.choices(PRIORITIES, weights=[1, 3, 7, 10], k=1)[0]
if state == "Failed":
failures += 1
if state == "Late":
late += 1
total_cost += cost
total_latency += duration
total_revenue += revenue
if d >= TODAY - timedelta(days=10) and (state in {"Failed", "Late"} or random.random() < 0.05):
runs.append({
"run_id": f"{d.strftime('%m%d')}-{region[:2]}-{len(runs)+1:04d}",
"date": d.strftime("%Y-%m-%d"),
"pipeline": pipeline,
"owner": owner,
"region": region,
"state": state,
"priority": priority,
"duration_s": duration,
"cost_usd": cost,
"revenue_k": revenue,
"sla_gap": round(max(0, duration - 120) / 60, 1),
})
daily_region_rows.append({
"date": d.strftime("%b %d"),
"region": region,
"runs": volume,
"failures": failures,
"late": late,
"success_rate": round(100 * (volume - failures - late * 0.35) / volume, 1),
"avg_latency": round(total_latency / volume, 1),
"cost_usd": round(total_cost, 2),
"revenue_k": round(total_revenue, 1),
})
runs = sorted(
runs,
key=lambda r: (r["priority"], r["state"] != "Failed", -r["duration_s"])
)[:80]
def aggregate_daily(rows):
by_date = defaultdict(lambda: {
"date": "",
"runs": 0,
"failures": 0,
"late": 0,
"cost_usd": 0.0,
"revenue_k": 0.0,
"latency_weighted": 0.0,
})
for r in rows:
bucket = by_date[r["date"]]
bucket["date"] = r["date"]
bucket["runs"] += r["runs"]
bucket["failures"] += r["failures"]
bucket["late"] += r["late"]
bucket["cost_usd"] += r["cost_usd"]
bucket["revenue_k"] += r["revenue_k"]
bucket["latency_weighted"] += r["avg_latency"] * r["runs"]
out = []
for d in [x.strftime("%b %d") for x in DATES]:
b = by_date[d]
if b["runs"]:
b["success_rate"] = round(100 * (b["runs"] - b["failures"] - b["late"] * 0.35) / b["runs"], 1)
b["avg_latency"] = round(b["latency_weighted"] / b["runs"], 1)
b["cost_usd"] = round(b["cost_usd"], 2)
b["revenue_k"] = round(b["revenue_k"], 1)
del b["latency_weighted"]
out.append(dict(b))
return out
def aggregate_regions(rows):
by_region = defaultdict(lambda: {
"region": "",
"runs": 0,
"failures": 0,
"late": 0,
"cost_usd": 0.0,
"revenue_k": 0.0,
"latency_weighted": 0.0,
})
for r in rows:
b = by_region[r["region"]]
b["region"] = r["region"]
b["runs"] += r["runs"]
b["failures"] += r["failures"]
b["late"] += r["late"]
b["cost_usd"] += r["cost_usd"]
b["revenue_k"] += r["revenue_k"]
b["latency_weighted"] += r["avg_latency"] * r["runs"]
out = []
for region in REGIONS[1:]:
b = by_region[region]
b["success_rate"] = round(100 * (b["runs"] - b["failures"] - b["late"] * 0.35) / b["runs"], 1)
b["avg_latency"] = round(b["latency_weighted"] / b["runs"], 1)
b["cost_usd"] = round(b["cost_usd"], 2)
b["revenue_k"] = round(b["revenue_k"], 1)
b["roi"] = round(b["revenue_k"] / max(1, b["cost_usd"]), 1)
del b["latency_weighted"]
out.append(dict(b))
return out
def make_status_rows(table_rows):
counts = Counter(r["state"] for r in table_rows)
return [{"state": k, "count": v} for k, v in counts.items()]
def make_pipeline_rows(table_rows):
counts = Counter(r["pipeline"] for r in table_rows)
return [{"pipeline": k, "count": v} for k, v in counts.most_common()]
def make_kpis(region, daily_rows, table_rows):
runs_count = sum(r["runs"] for r in daily_rows)
failures = sum(r["failures"] for r in daily_rows)
late = sum(r["late"] for r in daily_rows)
cost = sum(r["cost_usd"] for r in daily_rows)
revenue = sum(r["revenue_k"] for r in daily_rows)
return {
"region": region,
"runs": runs_count,
"success_rate": round(100 * (runs_count - failures - late * 0.35) / max(1, runs_count), 1),
"avg_latency": round(sum(r["avg_latency"] * r["runs"] for r in daily_rows) / max(1, runs_count), 1),
"cost_usd": round(cost, 2),
"revenue_k": round(revenue, 1),
"roi": round(revenue / max(1, cost), 1),
"open_issues": len(table_rows),
"p0p1": sum(1 for r in table_rows if r["priority"] in {"P0", "P1"}),
"failure_rate": round(100 * failures / max(1, runs_count), 2),
"spark": [r["success_rate"] for r in daily_rows[-14:]],
}
DAILY_BY_REGION = {"All": aggregate_daily(daily_region_rows)}
REGION_ROWS = aggregate_regions(daily_region_rows)
for region in REGIONS[1:]:
DAILY_BY_REGION[region] = [r for r in daily_region_rows if r["region"] == region]
RUNS_BY_REGION = {
region: [r for r in runs if region == "All" or r["region"] == region]
for region in REGIONS
}
STATUS_BY_REGION = {
region: make_status_rows(RUNS_BY_REGION[region])
for region in REGIONS
}
PIPELINE_BY_REGION = {
region: make_pipeline_rows(RUNS_BY_REGION[region])
for region in REGIONS
}
KPI_BY_REGION = {
region: make_kpis(region, DAILY_BY_REGION[region], RUNS_BY_REGION[region])
for region in REGIONS
}
WATCHLIST = sorted(
runs,
key=lambda r: (r["priority"], r["state"] != "Failed", -r["sla_gap"])
)[:8]
SCATTER_ROWS = [
{
"region": r["region"],
"success_rate": r["success_rate"],
"cost_usd": r["cost_usd"],
"avg_latency": r["avg_latency"],
}
for r in REGION_ROWS
]
RADAR_ROWS = [
{"metric": "Success", **{r["region"]: r["success_rate"] for r in REGION_ROWS}},
{"metric": "ROI", **{r["region"]: min(100, r["roi"] * 8) for r in REGION_ROWS}},
{"metric": "Latency", **{r["region"]: max(0, 100 - r["avg_latency"] / 2) for r in REGION_ROWS}},
{"metric": "Cost", **{r["region"]: max(0, 100 - r["cost_usd"] / 20) for r in REGION_ROWS}},
]
REGION_ACTIONS = {
region: [
SetState("selected_region", region),
SetState("line_rows", DAILY_BY_REGION[region]),
SetState("table_rows", RUNS_BY_REGION[region]),
SetState("status_rows", STATUS_BY_REGION[region]),
SetState("pipeline_rows", PIPELINE_BY_REGION[region]),
SetState("region_kpis", KPI_BY_REGION[region]),
SetState("selected_run", None),
ShowToast(f"Region set to {region}", variant="info", duration=1800),
]
for region in REGIONS
}
'''
Prefab コンポーネント、アクション、チャート、制御フロー、リアクティブ状態処理のための主要なインポートを追加します。地域、パイプライン、所有者、ステータス、優先度、コスト、レイテンシ、収益にわたって現実的な合成オペレーションデータを生成します。また、ダッシュボードのチャート、メトリクス、フィルター、テーブル、ウォッチリストを駆動する集計関数を作成し、データセットを準備します。
Copy CodeCopiedUse a different Browser
APP_CODE += r'''
Overview タブの構築
Copy CodeCopiedUse a different Browser
initial_state = {
"selected_region": "All",
"line_rows": DAILY_BY_REGION["All"],
"table_rows": RUNS_BY_REGION["All"],
"status_rows": STATUS_BY_REGION["All"],
"pipeline_rows": PIPELINE_BY_REGION["All"],
"region_kpis": KPI_BY_REGION["All"],
"selected_run": None,
"slo_target": 94,
"operator_name": "Colab Builder",
"notes": [{"note": "Click a run row to inspect it, then add triage notes here."}],
"watchlist": WATCHLIST,
"compact": False,
"dark_mode": False,
}
with PrefabApp(
title="Prefab Advanced Colab Tutorial",
state=initial_state,
css_class="max-w-7xl mx-auto p-6",
) as app:
with Column(gap=6):
with Row(justify="between", align="center", gap=4):
with Column(gap=1):
H2("Prefab Advanced Operations Dashboard")
Muted(
"A complete Colab-ready tutorial: Python DSL, reactive state, "
"charts, tables, forms, conditionals, actions, Mermaid, and static export."
)
with Row(gap=2, align="center"):
Badge(f"Region: {STATE.selected_region}", variant="info")
Badge(f"Operator: {STATE.operator_name}", variant="secondary")
Button(
"Docs",
variant="outline",
onClick=OpenLink("https://prefab.prefect.io/docs/welcome"),
)
with Alert(variant="info", icon="sparkles"):
AlertTitle("What this app demonstrates")
AlertDescription(
"Everything below is composed in Python. The exported HTML runs client-side "
"with no backend, while actions update Prefab state instantly."
)
with Card():
with CardHeader():
CardTitle("Interactive controls")
CardDescription(
"Use state actions to swap datasets, adjust SLO targets, and personalize "
"the UI without JavaScript."
)
with CardContent():
with Grid(columns={"sm": 1, "md": 2, "lg": 4}, gap=4):
with Column(gap=2):
Small("Region quick filters")
with Row(gap=2):
for region in REGIONS:
Button(
region,
size="sm",
variant="secondary" if region != "All" else "default",
onClick=REGION_ACTIONS[region],
)
with Column(gap=2):
Small("SLO target")
Slider(
value=STATE.slo_target,
min=80,
max=99,
step=0.5,
onChange=SetState("slo_target", EVENT),
gradient=True,
)
Muted(f"Target is {STATE.slo_target}% successful runs")
with Column(gap=2):
Small("Operator name")
Input(
value=STATE.operator_name,
placeholder="Your name",
on_change=SetState("operator_name", EVENT),
)
with Column(gap=2):
Small("Client-side toggles")
Switch(
value=STATE.compact,
label="Compact review mode",
onChange=ToggleState("compact"),
)
Switch(
value=STATE.dark_mode,
label="Dark-mode flag",
onChange=ToggleState("dark_mode"),
)
with Tabs(value="overview"):
with Tab("Overview", value="overview"):
with Column(gap=5):
with Grid(columns={"sm": 1, "md": 2, "lg": 4}, gap=4):
Metric(
label="Runs",
value=STATE.region_kpis.runs,
description=f"Selected slice: {STATE.selected_region}",
)
Metric(
label="Success rate",
value=f"{STATE.region_kpis.success_rate}%",
delta=f"Target {STATE.slo_target}%",
trend="up",
trendSentiment="positive",
)
Metric(
label="Avg latency",
value=f"{STATE.region_kpis.avg_latency}s",
description="Weighted 30-day average",
)
Metric(
label="ROI index",
value=STATE.region_kpis.roi,
description="Revenue thousands per USD cost",
)
with Grid(columns={"sm": 1, "lg": 3}, gap=4):
with Card(css_class="lg:col-span-2"):
with CardHeader():
CardTitle("Reliability trend")
CardDescription(
"Line chart bound to a reactive state key. "
"Region buttons replace the whole data list."
)
with CardContent():
LineChart(
data=STATE.line_rows,
xAxis="date",
series=[
ChartSeries(dataKey="success_rate", label="Success %"),
ChartSeries(dataKey="avg_latency", label="Avg latency"),
],
height=330,
curve="smooth",
showDots=False,
showLegend=True,
)
with Card():
with CardHeader():
CardTitle("SLO health")
CardDescription(
"Progress and ring components read live KPI and target state."
)
with CardContent():
with Column(gap=4, align="center"):
Ring(
value=STATE.region_kpis.success_rate,
target=STATE.slo_target,
label=f"{STATE.region_kpis.success_rate}%",
size="lg",
gradient=True,
)
Progress(
value=STATE.region_kpis.success_rate,
target=STATE.slo_target,
max=100,
gradient=True,
)
Sparkline(
data=STATE.region_kpis.spark,
height=56,
curve="smooth",
fill=True,
)
with If(STATE.region_kpis.success_rate >= STATE.slo_target):
Badge("Meeting target", variant="success")
with Else():
Badge("Below target", variant="warning")
with Grid(columns={"sm": 1, "lg": 2}, gap=4):
with Card():
with CardHeader():
CardTitle("Open issue status mix")
with CardContent():
PieChart(
data=STATE.status_rows,
dataKey="count",
nameKey="state",
原文を表示
In this tutorial, we build a Prefab application that demonstrates how to create interactive dashboards entirely in Python. We use Prefab’s component-based Python interface to design a polished operations dashboard with reactive state, charts, tables, filters, forms, tabs, alerts, metrics, and client-side actions. We generate realistic pipeline monitoring data, connect it to live UI controls, and export the final app as a static HTML dashboard that we can preview directly inside Google Colab. Through this workflow, we learn how Prefab lets us move from Python data logic to a modern React-powered user interface without having to write frontend code manually.
Installing Prefab in Colab
Copy CodeCopiedUse a different Browser
import os
import sys
import base64
import subprocess
from pathlib import Path
from IPython.display import HTML, display, FileLink
PREFAB_VERSION = "0.20.2"
APP_PATH = Path("/content/prefab_advanced_tutorial_app.py")
HTML_PATH = Path("/content/prefab_advanced_dashboard.html")
subprocess.check_call([
sys.executable,
"-m",
"pip",
"install",
"-q",
f"prefab-ui=={PREFAB_VERSION}",
])
APP_CODE = ""
We set up the Colab environment by importing the required Python utilities and defining the Prefab version, app path, and HTML export path. We install the pinned prefab-ui package so that the tutorial runs consistently without version-related issues. We also initialize an empty APP_CODE string, which we use to build the complete Prefab application step by step.
Copy CodeCopiedUse a different Browser
APP_CODE += r'''
Generating Synthetic Operations Data
Copy CodeCopiedUse a different Browser
import random
from collections import Counter, defaultdict
from datetime import date, timedelta
from prefab_ui.actions import AppendState, OpenLink, PopState, SetState, ShowToast, ToggleState
from prefab_ui.app import PrefabApp
from prefab_ui.components import (
Alert, AlertDescription, AlertTitle, Badge, Button, Card, CardContent,
CardDescription, CardFooter, CardHeader, CardTitle, Code, Column,
DataTable, DataTableColumn, Form, Grid, H2, Input, Markdown, Mermaid,
Metric, Muted, Progress, Ring, Row, Slider, Small, Switch, Tab, Tabs,
Text
)
from prefab_ui.components.charts import (
BarChart, ChartSeries, LineChart, PieChart, RadarChart, ScatterChart,
Sparkline
)
from prefab_ui.components.control_flow import Else, ForEach, If
from prefab_ui.rx import EVENT, STATE
random.seed(42)
TODAY = date.today()
DATES = [TODAY - timedelta(days=29 - i) for i in range(30)]
REGIONS = ["All", "APAC", "EMEA", "NA", "LATAM"]
PIPELINES = [
"Customer 360 ETL",
"Invoice OCR",
"LLM Triage",
"Risk Scoring",
"Forecast Sync",
"Warehouse Load",
]
OWNERS = ["Data Platform", "AI Apps", "Revenue Ops", "Risk Engineering"]
STATES = ["Completed", "Completed", "Completed", "Completed", "Late", "Failed"]
PRIORITIES = ["P0", "P1", "P2", "P3"]
runs = []
daily_region_rows = []
for d in DATES:
for region in REGIONS[1:]:
region_bias = {
"APAC": 0.96,
"EMEA": 0.94,
"NA": 0.97,
"LATAM": 0.91,
}[region]
volume = random.randint(32, 78)
failures = 0
late = 0
total_cost = 0.0
total_latency = 0.0
total_revenue = 0.0
for i in range(volume):
pipeline = random.choice(PIPELINES)
owner = random.choice(OWNERS)
state = random.choices(
STATES,
weights=[
region_bias * 10,
6,
4,
3,
1.2,
max(0.2, (1 - region_bias) * 16),
],
k=1,
)[0]
duration = max(
12,
int(
random.gauss(95, 35)
+ (20 if state == "Late" else 0)
+ (45 if state == "Failed" else 0)
),
)
cost = round(max(0.09, random.lognormvariate(-1.15, 0.55) + duration / 1800), 2)
revenue = round(random.uniform(1.2, 8.5) * (1.3 if state == "Completed" else 0.6), 2)
priority = random.choices(PRIORITIES, weights=[1, 3, 7, 10], k=1)[0]
if state == "Failed":
failures += 1
if state == "Late":
late += 1
total_cost += cost
total_latency += duration
total_revenue += revenue
if d >= TODAY - timedelta(days=10) and (state in {"Failed", "Late"} or random.random() < 0.05):
runs.append({
"run_id": f"{d.strftime('%m%d')}-{region[:2]}-{len(runs)+1:04d}",
"date": d.strftime("%Y-%m-%d"),
"pipeline": pipeline,
"owner": owner,
"region": region,
"state": state,
"priority": priority,
"duration_s": duration,
"cost_usd": cost,
"revenue_k": revenue,
"sla_gap": round(max(0, duration - 120) / 60, 1),
})
daily_region_rows.append({
"date": d.strftime("%b %d"),
"region": region,
"runs": volume,
"failures": failures,
"late": late,
"success_rate": round(100 * (volume - failures - late * 0.35) / volume, 1),
"avg_latency": round(total_latency / volume, 1),
"cost_usd": round(total_cost, 2),
"revenue_k": round(total_revenue, 1),
})
runs = sorted(
runs,
key=lambda r: (r["priority"], r["state"] != "Failed", -r["duration_s"])
)[:80]
def aggregate_daily(rows):
by_date = defaultdict(lambda: {
"date": "",
"runs": 0,
"failures": 0,
"late": 0,
"cost_usd": 0.0,
"revenue_k": 0.0,
"latency_weighted": 0.0,
})
for r in rows:
bucket = by_date[r["date"]]
bucket["date"] = r["date"]
bucket["runs"] += r["runs"]
bucket["failures"] += r["failures"]
bucket["late"] += r["late"]
bucket["cost_usd"] += r["cost_usd"]
bucket["revenue_k"] += r["revenue_k"]
bucket["latency_weighted"] += r["avg_latency"] * r["runs"]
out = []
for d in [x.strftime("%b %d") for x in DATES]:
b = by_date[d]
if b["runs"]:
b["success_rate"] = round(100 * (b["runs"] - b["failures"] - b["late"] * 0.35) / b["runs"], 1)
b["avg_latency"] = round(b["latency_weighted"] / b["runs"], 1)
b["cost_usd"] = round(b["cost_usd"], 2)
b["revenue_k"] = round(b["revenue_k"], 1)
del b["latency_weighted"]
out.append(dict(b))
return out
def aggregate_regions(rows):
by_region = defaultdict(lambda: {
"region": "",
"runs": 0,
"failures": 0,
"late": 0,
"cost_usd": 0.0,
"revenue_k": 0.0,
"latency_weighted": 0.0,
})
for r in rows:
b = by_region[r["region"]]
b["region"] = r["region"]
b["runs"] += r["runs"]
b["failures"] += r["failures"]
b["late"] += r["late"]
b["cost_usd"] += r["cost_usd"]
b["revenue_k"] += r["revenue_k"]
b["latency_weighted"] += r["avg_latency"] * r["runs"]
out = []
for region in REGIONS[1:]:
b = by_region[region]
b["success_rate"] = round(100 * (b["runs"] - b["failures"] - b["late"] * 0.35) / b["runs"], 1)
b["avg_latency"] = round(b["latency_weighted"] / b["runs"], 1)
b["cost_usd"] = round(b["cost_usd"], 2)
b["revenue_k"] = round(b["revenue_k"], 1)
b["roi"] = round(b["revenue_k"] / max(1, b["cost_usd"]), 1)
del b["latency_weighted"]
out.append(dict(b))
return out
def make_status_rows(table_rows):
counts = Counter(r["state"] for r in table_rows)
return [{"state": k, "count": v} for k, v in counts.items()]
def make_pipeline_rows(table_rows):
counts = Counter(r["pipeline"] for r in table_rows)
return [{"pipeline": k, "count": v} for k, v in counts.most_common()]
def make_kpis(region, daily_rows, table_rows):
runs_count = sum(r["runs"] for r in daily_rows)
failures = sum(r["failures"] for r in daily_rows)
late = sum(r["late"] for r in daily_rows)
cost = sum(r["cost_usd"] for r in daily_rows)
revenue = sum(r["revenue_k"] for r in daily_rows)
return {
"region": region,
"runs": runs_count,
"success_rate": round(100 * (runs_count - failures - late * 0.35) / max(1, runs_count), 1),
"avg_latency": round(sum(r["avg_latency"] * r["runs"] for r in daily_rows) / max(1, runs_count), 1),
"cost_usd": round(cost, 2),
"revenue_k": round(revenue, 1),
"roi": round(revenue / max(1, cost), 1),
"open_issues": len(table_rows),
"p0p1": sum(1 for r in table_rows if r["priority"] in {"P0", "P1"}),
"failure_rate": round(100 * failures / max(1, runs_count), 2),
"spark": [r["success_rate"] for r in daily_rows[-14:]],
}
DAILY_BY_REGION = {"All": aggregate_daily(daily_region_rows)}
REGION_ROWS = aggregate_regions(daily_region_rows)
for region in REGIONS[1:]:
DAILY_BY_REGION[region] = [r for r in daily_region_rows if r["region"] == region]
RUNS_BY_REGION = {
region: [r for r in runs if region == "All" or r["region"] == region]
for region in REGIONS
}
STATUS_BY_REGION = {
region: make_status_rows(RUNS_BY_REGION[region])
for region in REGIONS
}
PIPELINE_BY_REGION = {
region: make_pipeline_rows(RUNS_BY_REGION[region])
for region in REGIONS
}
KPI_BY_REGION = {
region: make_kpis(region, DAILY_BY_REGION[region], RUNS_BY_REGION[region])
for region in REGIONS
}
WATCHLIST = sorted(
runs,
key=lambda r: (r["priority"], r["state"] != "Failed", -r["sla_gap"])
)[:8]
SCATTER_ROWS = [
{
"region": r["region"],
"success_rate": r["success_rate"],
"cost_usd": r["cost_usd"],
"avg_latency": r["avg_latency"],
}
for r in REGION_ROWS
]
RADAR_ROWS = [
{"metric": "Success", **{r["region"]: r["success_rate"] for r in REGION_ROWS}},
{"metric": "ROI", **{r["region"]: min(100, r["roi"] * 8) for r in REGION_ROWS}},
{"metric": "Latency", **{r["region"]: max(0, 100 - r["avg_latency"] / 2) for r in REGION_ROWS}},
{"metric": "Cost", **{r["region"]: max(0, 100 - r["cost_usd"] / 20) for r in REGION_ROWS}},
]
REGION_ACTIONS = {
region: [
SetState("selected_region", region),
SetState("line_rows", DAILY_BY_REGION[region]),
SetState("table_rows", RUNS_BY_REGION[region]),
SetState("status_rows", STATUS_BY_REGION[region]),
SetState("pipeline_rows", PIPELINE_BY_REGION[region]),
SetState("region_kpis", KPI_BY_REGION[region]),
SetState("selected_run", None),
ShowToast(f"Region set to {region}", variant="info", duration=1800),
]
for region in REGIONS
}
'''
We add the main imports for Prefab components, actions, charts, control flow, and reactive state handling. We generate realistic synthetic operations data across regions, pipelines, owners, states, priorities, costs, latency, and revenue. We also create aggregation functions and prepare datasets that power the dashboard’s charts, metrics, filters, tables, and watchlist.
Copy CodeCopiedUse a different Browser
APP_CODE += r'''
Building the Overview Tab
Copy CodeCopiedUse a different Browser
initial_state = {
"selected_region": "All",
"line_rows": DAILY_BY_REGION["All"],
"table_rows": RUNS_BY_REGION["All"],
"status_rows": STATUS_BY_REGION["All"],
"pipeline_rows": PIPELINE_BY_REGION["All"],
"region_kpis": KPI_BY_REGION["All"],
"selected_run": None,
"slo_target": 94,
"operator_name": "Colab Builder",
"notes": [{"note": "Click a run row to inspect it, then add triage notes here."}],
"watchlist": WATCHLIST,
"compact": False,
"dark_mode": False,
}
with PrefabApp(
title="Prefab Advanced Colab Tutorial",
state=initial_state,
css_class="max-w-7xl mx-auto p-6",
) as app:
with Column(gap=6):
with Row(justify="between", align="center", gap=4):
with Column(gap=1):
H2("Prefab Advanced Operations Dashboard")
Muted(
"A complete Colab-ready tutorial: Python DSL, reactive state, "
"charts, tables, forms, conditionals, actions, Mermaid, and static export."
)
with Row(gap=2, align="center"):
Badge(f"Region: {STATE.selected_region}", variant="info")
Badge(f"Operator: {STATE.operator_name}", variant="secondary")
Button(
"Docs",
variant="outline",
onClick=OpenLink("https://prefab.prefect.io/docs/welcome"),
)
with Alert(variant="info", icon="sparkles"):
AlertTitle("What this app demonstrates")
AlertDescription(
"Everything below is composed in Python. The exported HTML runs client-side "
"with no backend, while actions update Prefab state instantly."
)
with Card():
with CardHeader():
CardTitle("Interactive controls")
CardDescription(
"Use state actions to swap datasets, adjust SLO targets, and personalize "
"the UI without JavaScript."
)
with CardContent():
with Grid(columns={"sm": 1, "md": 2, "lg": 4}, gap=4):
with Column(gap=2):
Small("Region quick filters")
with Row(gap=2):
for region in REGIONS:
Button(
region,
size="sm",
variant="secondary" if region != "All" else "default",
onClick=REGION_ACTIONS[region],
)
with Column(gap=2):
Small("SLO target")
Slider(
value=STATE.slo_target,
min=80,
max=99,
step=0.5,
onChange=SetState("slo_target", EVENT),
gradient=True,
)
Muted(f"Target is {STATE.slo_target}% successful runs")
with Column(gap=2):
Small("Operator name")
Input(
value=STATE.operator_name,
placeholder="Your name",
on_change=SetState("operator_name", EVENT),
)
with Column(gap=2):
Small("Client-side toggles")
Switch(
value=STATE.compact,
label="Compact review mode",
onChange=ToggleState("compact"),
)
Switch(
value=STATE.dark_mode,
label="Dark-mode flag",
onChange=ToggleState("dark_mode"),
)
with Tabs(value="overview"):
with Tab("Overview", value="overview"):
with Column(gap=5):
with Grid(columns={"sm": 1, "md": 2, "lg": 4}, gap=4):
Metric(
label="Runs",
value=STATE.region_kpis.runs,
description=f"Selected slice: {STATE.selected_region}",
)
Metric(
label="Success rate",
value=f"{STATE.region_kpis.success_rate}%",
delta=f"Target {STATE.slo_target}%",
trend="up",
trendSentiment="positive",
)
Metric(
label="Avg latency",
value=f"{STATE.region_kpis.avg_latency}s",
description="Weighted 30-day average",
)
Metric(
label="ROI index",
value=STATE.region_kpis.roi,
description="Revenue thousands per USD cost",
)
with Grid(columns={"sm": 1, "lg": 3}, gap=4):
with Card(css_class="lg:col-span-2"):
with CardHeader():
CardTitle("Reliability trend")
CardDescription(
"Line chart bound to a reactive state key. "
"Region buttons replace the whole data list."
)
with CardContent():
LineChart(
data=STATE.line_rows,
xAxis="date",
series=[
ChartSeries(dataKey="success_rate", label="Success %"),
ChartSeries(dataKey="avg_latency", label="Avg latency"),
],
height=330,
curve="smooth",
showDots=False,
showLegend=True,
)
with Card():
with CardHeader():
CardTitle("SLO health")
CardDescription(
"Progress and ring components read live KPI and target state."
)
with CardContent():
with Column(gap=4, align="center"):
Ring(
value=STATE.region_kpis.success_rate,
target=STATE.slo_target,
label=f"{STATE.region_kpis.success_rate}%",
size="lg",
gradient=True,
)
Progress(
value=STATE.region_kpis.success_rate,
target=STATE.slo_target,
max=100,
gradient=True,
)
Sparkline(
data=STATE.region_kpis.spark,
height=56,
curve="smooth",
fill=True,
)
with If(STATE.region_kpis.success_rate >= STATE.slo_target):
Badge("Meeting target", variant="success")
with Else():
Badge("Below target", variant="warning")
with Grid(columns={"sm": 1, "lg": 2}, gap=4):
with Card():
with CardHeader():
CardTitle("Open issue status mix")
with CardContent():
PieChart(
data=STATE.status_rows,
dataKey="count",
nameKey="state",
関連記事
Simon Willison Blog の datasette-tailscale 0.1a0 リリース
Simon Willison が、Datasette サーバーを Tailscale ネットワークに接続する実験的プラグイン「datasette-tailscale」のバージョン 0.1a0 を公開した。これにより、認証キーとホスト名を指定してローカルサーバーから安全にデータへアクセスできるようになる。
Graphify と NetworkX を用いた Python コードベース構造の可視化：ゴッドノード、コミュニティ、アーキテクチャ図の作成
MarkTechPost は、Graphify ツールと NetworkX ライブラリを使用して、Python アプリケーションを知識グラフに変換し、オフラインでコード構造を可視化するチュートリアルを紹介している。
データセット 1.0a35 のリリース
Simon Willison が、データベース操作メニューに「テーブル作成」インターフェースを追加した大規模な新バージョン「datasette 1.0a35」を公開しました。
今日のまとめ
AI日報で今日の重要ニュースをまとめ読み