Prefab のリアクティブ UI コンポーネントと静的 HTML エクスポートを活用した Python フォーカスのインタラクティブダッシュボード設計方法
本文の状態
日本語全文を表示中
詳細モードで約9分の本文を読めます。
同じ出来事の情報源
この情報源を基点に整理
MarkTechPost
MarkTechPost は、Prefab のコンポーネントベースの Python インターフェースを使用して、リアクティブな状態管理やチャート機能を持つ操作用ダッシュボードを構築し、最終的に Google Colab で直接プレビュー可能な静的 HTML としてエクスポートする手法を紹介している。
Continue in AI NEW LAB
このニュースを、実務の判断につなげる
AI NEW LABで、試したことや先に確認したい条件を共有できます。まずはログインなしで読めます。
AI NEW LABで論点を見るSource Article
元記事を日本語で読む
本文に関係しない購読案内、埋め込み通知、サイト内プロモーションは除いています。
このチュートリアルでは、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",
関連記事
今日のまとめ
AIデイリーブリーフで今日の重要ニュースをまとめ読み