Run Class

Run object for tracking experiments.

dalva.Run

Run(
    project: str,
    name: str | None = None,
    config: dict | None = None,
    resume: str | None = None,
    server_url: str = "http://localhost:8000",
)

Run object for tracking experiments via HTTP.

All operations are performed via HTTP requests to the server. No local database operations are performed.

Example

run = Run(project="my-project", config={"lr": 0.001})
run.log({"loss": 0.5}, step=0)
run.finish()

Initialize a run by creating it on the server.

Parameters:

Name	Type	Description	Default
project	str	Project name	required
name	str | None	Optional run name (user-defined, for display only)	None
config	dict | None	Optional configuration dictionary	None
resume	str | None	run_id to resume (omit to create a new run)	None
server_url	str	Server URL. Defaults to http://localhost:8000	'http://localhost:8000'

Source code in backend/src/dalva/run.py

def __init__(
    self,
    project: str,
    name: str | None = None,
    config: dict | None = None,
    resume: str | None = None,
    server_url: str = "http://localhost:8000",
):
    """
    Initialize a run by creating it on the server.

    Args:
        project: Project name
        name: Optional run name (user-defined, for display only)
        config: Optional configuration dictionary
        resume: run_id to resume (omit to create a new run)
        server_url: Server URL. Defaults to http://localhost:8000
    """
    self.project_name = project
    self.config = config or {}
    self._server_url = server_url
    self._client: httpx.Client | None = None

    # Verify server is accessible
    self._verify_server_connection()

    # Create run via API
    self._create_run_on_server(name=name, config=self.config, resume=resume)

    # Print run ID for user convenience
    print(f"Run created: {self.run_id}")

log

log(metrics: dict[str, Any], step: int | None = None)

Log metrics to the run.

Parameters:

Name	Type	Description	Default
metrics	dict[str, Any]	Dictionary of metric name -> value	required
step	int | None	Optional step number for series values	None

Example

run = Run(project="my-project", config={"lr": 0.001})
run.log({"accuracy": 0.85})
for step in range(100):
    run.log({"loss": 0.5, "accuracy": 0.5}, step=step)
run.finish()

Source code in backend/src/dalva/run.py

def log(self, metrics: dict[str, Any], step: int | None = None):
    """Log metrics to the run.

    Args:
        metrics: Dictionary of metric name -> value
        step: Optional step number for series values

    Example:
        ```python
        run = Run(project="my-project", config={"lr": 0.001})
        run.log({"accuracy": 0.85})
        for step in range(100):
            run.log({"loss": 0.5, "accuracy": 0.5}, step=step)
        run.finish()
        ```
    """
    client = self._get_client()
    payload = {
        "metrics": metrics,
        "step": step,
    }
    try:
        response = client.post(f"/api/runs/{self._db_id}/log", json=payload)
        response.raise_for_status()
    except httpx.HTTPError as e:
        raise ConnectionError(f"Failed to log metrics to server: {e}")

finish

finish()

Finish the run and mark it as completed.

Example

run = Run(project="my-project")
run.log({"loss": 0.5}, step=0)
run.finish()

Source code in backend/src/dalva/run.py

def finish(self):
    """Finish the run and mark it as completed.

    Example:
        ```python
        run = Run(project="my-project")
        run.log({"loss": 0.5}, step=0)
        run.finish()
        ```
    """
    client = self._get_client()
    try:
        response = client.post(f"/api/runs/{self._db_id}/finish")
        response.raise_for_status()
        result = response.json()
        print(f"[Run] Run finished (state={result['state']})")
    except httpx.HTTPError as e:
        raise ConnectionError(f"Failed to finish run on server: {e}")
    finally:
        self._client = None