DataEngineX dashboard — Streamlit-based pipeline monitoring.
Provides a configurable dashboard template that pulls data from
Prometheus metrics and displays pipeline status, quality scores,
model drift, and alerts.
Requires streamlit (install via the dashboard dependency group).
Usage::
from dataenginex.dashboard import DashboardConfig, BaseDashboard
config = DashboardConfig.from_yaml("dashboard_config.yaml")
dashboard = BaseDashboard(config)
dashboard.run()
Or standalone::
streamlit run examples/dashboard/run_dashboard.py
BaseDashboard
Streamlit-based dashboard for DataEngineX pipeline monitoring.
Usage::
config = DashboardConfig.from_yaml("config.yaml")
dashboard = BaseDashboard(config)
dashboard.run()
Source code in src/dataenginex/dashboard/app.py
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165 | class BaseDashboard:
"""Streamlit-based dashboard for DataEngineX pipeline monitoring.
Usage::
config = DashboardConfig.from_yaml("config.yaml")
dashboard = BaseDashboard(config)
dashboard.run()
"""
def __init__(self, config: DashboardConfig | None = None) -> None:
self.config = config or DashboardConfig()
def run(self) -> None:
"""Render the dashboard using Streamlit.
Must be called from within ``streamlit run``.
"""
if not _HAS_STREAMLIT:
msg = (
"streamlit is required for the dashboard. "
"Install it with: uv sync --group dashboard"
)
raise ImportError(msg)
from dataenginex.dashboard.panels import (
alerts_panel,
model_drift_panel,
pipeline_status_panel,
quality_scores_panel,
)
st.set_page_config(
page_title=self.config.title,
page_icon="📊",
layout="wide",
)
st.title(self.config.title)
st.caption(f"Auto-refresh: {self.config.refresh_interval_seconds}s")
metrics = _fetch_metrics(self.config)
panel_map = {
"pipeline_status": pipeline_status_panel,
"quality_scores": quality_scores_panel,
"model_drift": model_drift_panel,
"alerts": alerts_panel,
}
for panel_name in self.config.panels:
fn = panel_map.get(panel_name)
if fn:
fn(metrics)
else:
st.warning(f"Unknown panel: {panel_name}")
st.divider()
st.caption(f"API: {self.config.dex_api_url} | Prometheus: {self.config.prometheus_url}")
|
run()
Render the dashboard using Streamlit.
Must be called from within streamlit run.
Source code in src/dataenginex/dashboard/app.py
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165 | def run(self) -> None:
"""Render the dashboard using Streamlit.
Must be called from within ``streamlit run``.
"""
if not _HAS_STREAMLIT:
msg = (
"streamlit is required for the dashboard. "
"Install it with: uv sync --group dashboard"
)
raise ImportError(msg)
from dataenginex.dashboard.panels import (
alerts_panel,
model_drift_panel,
pipeline_status_panel,
quality_scores_panel,
)
st.set_page_config(
page_title=self.config.title,
page_icon="📊",
layout="wide",
)
st.title(self.config.title)
st.caption(f"Auto-refresh: {self.config.refresh_interval_seconds}s")
metrics = _fetch_metrics(self.config)
panel_map = {
"pipeline_status": pipeline_status_panel,
"quality_scores": quality_scores_panel,
"model_drift": model_drift_panel,
"alerts": alerts_panel,
}
for panel_name in self.config.panels:
fn = panel_map.get(panel_name)
if fn:
fn(metrics)
else:
st.warning(f"Unknown panel: {panel_name}")
st.divider()
st.caption(f"API: {self.config.dex_api_url} | Prometheus: {self.config.prometheus_url}")
|
DashboardConfig
Bases: BaseModel
Dashboard configuration loaded from YAML.
Source code in src/dataenginex/dashboard/app.py
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61 | class DashboardConfig(BaseModel):
"""Dashboard configuration loaded from YAML."""
title: str = Field(default="DataEngineX Dashboard", description="Page title")
prometheus_url: str = Field(
default="http://localhost:9090",
description="Base URL of the Prometheus server",
)
dex_api_url: str = Field(
default="http://localhost:17000",
description="Base URL of the DEX API for metrics/health",
)
refresh_interval_seconds: int = Field(default=30, ge=5, description="Auto-refresh interval")
panels: list[str] = Field(
default_factory=lambda: [
"pipeline_status",
"quality_scores",
"model_drift",
"alerts",
],
description="Which panels to display",
)
@classmethod
def from_yaml(cls, path: str | Path) -> DashboardConfig:
"""Load configuration from a YAML file."""
p = Path(path)
if not p.exists():
logger.warning("config file not found, using defaults", path=str(p))
return cls()
raw = yaml.safe_load(p.read_text(encoding="utf-8"))
if not isinstance(raw, dict):
logger.warning("invalid config format, using defaults", path=str(p))
return cls()
return cls(**raw)
|
from_yaml(path)
classmethod
Load configuration from a YAML file.
Source code in src/dataenginex/dashboard/app.py
50
51
52
53
54
55
56
57
58
59
60
61 | @classmethod
def from_yaml(cls, path: str | Path) -> DashboardConfig:
"""Load configuration from a YAML file."""
p = Path(path)
if not p.exists():
logger.warning("config file not found, using defaults", path=str(p))
return cls()
raw = yaml.safe_load(p.read_text(encoding="utf-8"))
if not isinstance(raw, dict):
logger.warning("invalid config format, using defaults", path=str(p))
return cls()
return cls(**raw)
|