dlt._workspace.helpers.dashboard.utils.visualization
Pipeline execution timeline visualization and load package status badges.
badge_html
def badge_html(text: str, color: str, size: str = "small") -> str
Build a colored badge HTML element.
migration_badge
def migration_badge(count: int) -> str
Build migration badge HTML using CSS classes.
status_badge
def status_badge(status: TPipelineRunStatus) -> str
Build status badge HTML using CSS classes.
pipeline_execution_html
def pipeline_execution_html(transaction_id: str,
status: TPipelineRunStatus,
steps_data: List[PipelineStepData],
migrations_count: int = 0,
finished_at: Optional[datetime] = None) -> mo.Html
Build an HTML visualization for a pipeline execution using CSS classes.
get_steps_data_and_status
def get_steps_data_and_status(
trace_steps: List[PipelineStepTrace]
) -> Tuple[List[PipelineStepData], TPipelineRunStatus]
Get trace steps data and the status of the corresponding pipeline execution.
get_migrations_count
def get_migrations_count(last_load_info: LoadInfo) -> int
Count the number of unique migrations (schema versions) from load packages.
pipeline_execution_visualization
def pipeline_execution_visualization(
trace: PipelineTrace) -> Optional[mo.Html]
Create a visual timeline of pipeline run showing extract, normalize and load steps.
collect_load_packages_from_trace
def collect_load_packages_from_trace(
trace: PipelineTrace) -> List[LoadPackageInfo]
Collect all unique load packages from all steps.
load_package_status_labels
def load_package_status_labels(trace: PipelineTrace) -> mo.ui.table
Build a table of load package status badges for each package in the trace.
For each package, determines its visual status badge based on whether the package is partially loaded, pending, or in a final state.