MlflowRunTable

MlflowRunTable

MlflowRunTable(*, tracking_uri: str | None = None, experiment_id: str | None = None, poll_seconds: int | None = None, width: int = 900, **kwargs: Any)

Bases: AnyWidget

Interactive table showing MLflow experiment runs.

Displays run name, status, start time, duration, parameters, and final metric values in a sortable HTML table.

Examples:

from mlflow_widgets import MlflowRunTable

table = MlflowRunTable(
    tracking_uri="http://localhost:5000",
    experiment_id="1",
)
table

Create a MlflowRunTable widget.

Parameters:

Name	Type	Description	Default
tracking_uri	str | None	MLflow tracking server URI. Defaults to MLFLOW_TRACKING_URI env var or http://localhost:5000.	None
experiment_id	str | None	MLflow experiment ID to display runs for.	None
poll_seconds	int | None	Seconds between auto-refresh, or None to disable auto-polling and show a manual refresh button.	None
width	int	Table width in pixels.	900

Source code in src/mlflow_widgets/table.py

def __init__(
    self,
    *,
    tracking_uri: str | None = None,
    experiment_id: str | None = None,
    poll_seconds: int | None = None,
    width: int = 900,
    **kwargs: Any,
) -> None:
    """Create a MlflowRunTable widget.

    Args:
        tracking_uri: MLflow tracking server URI. Defaults to
            ``MLFLOW_TRACKING_URI`` env var or ``http://localhost:5000``.
        experiment_id: MLflow experiment ID to display runs for.
        poll_seconds: Seconds between auto-refresh, or ``None`` to
            disable auto-polling and show a manual refresh button.
        width: Table width in pixels.
    """
    self._tracking_uri = tracking_uri or os.environ.get(
        "MLFLOW_TRACKING_URI", "http://localhost:5000"
    )
    self._experiment_id = experiment_id
    self._mlflow = _try_import_mlflow()
    self._client = None
    if self._mlflow is not None:
        self._client = self._mlflow.MlflowClient(self._tracking_uri)

    self._poll_timer: threading.Timer | None = None
    self._poll_lock = threading.Lock()
    self._stopped = False

    super().__init__(poll_seconds=poll_seconds, width=width, **kwargs)

    if experiment_id:
        self.refresh()

    if poll_seconds is not None and poll_seconds > 0:
        self._schedule_poll()

refresh

refresh() -> None

Fetch run data from MLflow and update the table.

Source code in src/mlflow_widgets/table.py

def refresh(self) -> None:
    """Fetch run data from MLflow and update the table."""
    if not self._experiment_id:
        self._status = "No experiment_id set"
        return

    try:
        if self._client is not None:
            rows = _search_runs_detailed_client(self._client, self._experiment_id)
        else:
            rows = _search_runs_detailed_rest(self._tracking_uri, self._experiment_id)

        all_param_keys: set[str] = set()
        all_metric_keys: set[str] = set()
        for row in rows:
            all_param_keys.update(row["params"].keys())
            all_metric_keys.update(row["metrics"].keys())

        now = datetime.now().strftime("%H:%M:%S")
        self._table_data = rows
        self._param_keys = sorted(all_param_keys)
        self._metric_keys = sorted(all_metric_keys)
        self._status = f"{len(rows)} runs (updated {now})"
    except Exception as exc:
        self._status = f"Error: {exc}"

stop

stop() -> None

Stop auto-polling.

Source code in src/mlflow_widgets/table.py

def stop(self) -> None:
    """Stop auto-polling."""
    self._stopped = True
    with self._poll_lock:
        if self._poll_timer is not None:
            self._poll_timer.cancel()
            self._poll_timer = None

close

close() -> None

Source code in src/mlflow_widgets/table.py

def close(self) -> None:
    self.stop()
    super().close()