Index

routes

Dashboard API routes.

All routes are prefixed with /api for clear API namespace separation.

Classes

SheetSummary

Bases: BaseModel

Summarized sheet information for list views.

JobSummary

Bases: BaseModel

Summarized job information for list views.

Functions

from_checkpoint classmethod

from_checkpoint(state)

Create from CheckpointState.

Source code in src/marianne/dashboard/routes/__init__.py

@classmethod
def from_checkpoint(cls, state: CheckpointState) -> "JobSummary":
    """Create from CheckpointState."""
    completed, total = state.get_progress()
    return cls(
        job_id=state.job_id,
        job_name=state.job_name,
        status=state.status,
        total_sheets=total,
        completed_sheets=completed,
        progress_percent=state.get_progress_percent(),
        created_at=state.created_at,
        updated_at=state.updated_at,
    )

JobDetail

Bases: BaseModel

Full job details including sheet information.

Functions

from_checkpoint classmethod

from_checkpoint(state)

Create from CheckpointState.

Source code in src/marianne/dashboard/routes/__init__.py

@classmethod
def from_checkpoint(cls, state: CheckpointState) -> "JobDetail":
    """Create from CheckpointState."""
    sheets = [
        SheetSummary(
            sheet_num=s.sheet_num,
            status=s.status,
            attempt_count=s.attempt_count,
            validation_passed=s.validation_passed,
        )
        for s in sorted(state.sheets.values(), key=lambda x: x.sheet_num)
    ]
    return cls(
        job_id=state.job_id,
        job_name=state.job_name,
        status=state.status,
        total_sheets=state.total_sheets,
        last_completed_sheet=state.last_completed_sheet,
        current_sheet=state.current_sheet,
        progress_percent=state.get_progress_percent(),
        created_at=state.created_at,
        updated_at=state.updated_at,
        started_at=state.started_at,
        completed_at=state.completed_at,
        error_message=state.error_message,
        total_retry_count=state.total_retry_count,
        rate_limit_waits=state.rate_limit_waits,
        sheets=sheets,
    )

JobStatusResponse

Bases: BaseModel

Focused status information for job monitoring.

Functions

from_checkpoint classmethod

from_checkpoint(state)

Create from CheckpointState.

Source code in src/marianne/dashboard/routes/__init__.py

@classmethod
def from_checkpoint(cls, state: CheckpointState) -> "JobStatusResponse":
    """Create from CheckpointState."""
    completed, total = state.get_progress()
    return cls(
        job_id=state.job_id,
        status=state.status,
        progress_percent=state.get_progress_percent(),
        completed_sheets=completed,
        total_sheets=total,
        current_sheet=state.current_sheet,
        error_message=state.error_message,
        updated_at=state.updated_at,
    )

JobListResponse

Bases: BaseModel

Response for job list endpoint.

Functions

resolve_job_workspace

resolve_job_workspace(state, job_id)

Resolve workspace path from job state's worktree path.

Only works for worktree-isolated jobs. Non-isolated jobs do not store a workspace path in state, so this function will raise 404 for them.

Parameters:

Name	Type	Description	Default
state	CheckpointState	Loaded checkpoint state (must have worktree_path set).	required
job_id	str	Job identifier (for error messages).	required

Returns:

Type	Description
Path	Resolved workspace Path.

Raises:

Type	Description
HTTPException	404 if no worktree_path is set on the state.

Source code in src/marianne/dashboard/routes/__init__.py

def resolve_job_workspace(state: CheckpointState, job_id: str) -> Path:
    """Resolve workspace path from job state's worktree path.

    Only works for worktree-isolated jobs. Non-isolated jobs do not store
    a workspace path in state, so this function will raise 404 for them.

    Args:
        state: Loaded checkpoint state (must have worktree_path set).
        job_id: Job identifier (for error messages).

    Returns:
        Resolved workspace Path.

    Raises:
        HTTPException: 404 if no worktree_path is set on the state.
    """
    if state.worktree_path:
        return Path(state.worktree_path)
    raise HTTPException(
        status_code=404,
        detail=f"No accessible workspace found for job {job_id}. "
               f"Job may not be using worktree isolation.",
    )

get_job_or_404 async

get_job_or_404(backend, job_id)

Load job state or raise 404 if not found.

Consolidates the repeated load→check→raise pattern used across multiple route handlers.

Parameters:

Name	Type	Description	Default
backend	StateBackend	State backend to load from.	required
job_id	str	Job identifier.	required

Returns:

Type	Description
CheckpointState	Loaded CheckpointState.

Raises:

Type	Description
HTTPException	404 if job not found.

Source code in src/marianne/dashboard/routes/__init__.py

async def get_job_or_404(
    backend: StateBackend, job_id: str
) -> CheckpointState:
    """Load job state or raise 404 if not found.

    Consolidates the repeated load→check→raise pattern used
    across multiple route handlers.

    Args:
        backend: State backend to load from.
        job_id: Job identifier.

    Returns:
        Loaded CheckpointState.

    Raises:
        HTTPException: 404 if job not found.
    """
    state = await backend.load(job_id)
    if state is None:
        raise HTTPException(status_code=404, detail=f"Score not found: {job_id}")
    return state

list_jobs async

list_jobs(status=None, limit=50, backend=Depends(get_state_backend))

List all jobs with optional status filter.

Parameters:

Name	Type	Description	Default
status	JobStatus | None	Filter by job status (optional)	None
limit	int	Maximum number of jobs to return	50
backend	StateBackend	State backend (injected)	Depends(get_state_backend)

Returns:

Type	Description
JobListResponse	List of job summaries

Source code in src/marianne/dashboard/routes/__init__.py

@router.get("/jobs", response_model=JobListResponse)
async def list_jobs(
    status: JobStatus | None = None,
    limit: int = 50,
    backend: StateBackend = Depends(get_state_backend),
) -> JobListResponse:
    """List all jobs with optional status filter.

    Args:
        status: Filter by job status (optional)
        limit: Maximum number of jobs to return
        backend: State backend (injected)

    Returns:
        List of job summaries
    """
    all_jobs = await backend.list_jobs()

    # Apply status filter
    if status is not None:
        all_jobs = [j for j in all_jobs if j.status == status]

    # Apply limit
    limited_jobs = all_jobs[:limit]

    return JobListResponse(
        jobs=[JobSummary.from_checkpoint(j) for j in limited_jobs],
        total=len(all_jobs),
    )

get_job async

get_job(job_id, backend=Depends(get_state_backend))

Get detailed information about a specific job.

Parameters:

Name	Type	Description	Default
job_id	str	Unique job identifier	required
backend	StateBackend	State backend (injected)	Depends(get_state_backend)

Returns:

Type	Description
JobDetail	Full job details

Raises:

Type	Description
HTTPException	404 if job not found

Source code in src/marianne/dashboard/routes/__init__.py

@router.get("/jobs/{job_id}", response_model=JobDetail)
async def get_job(
    job_id: str,
    backend: StateBackend = Depends(get_state_backend),
) -> JobDetail:
    """Get detailed information about a specific job.

    Args:
        job_id: Unique job identifier
        backend: State backend (injected)

    Returns:
        Full job details

    Raises:
        HTTPException: 404 if job not found
    """
    state = await get_job_or_404(backend, job_id)
    return JobDetail.from_checkpoint(state)

get_job_status async

get_job_status(job_id, backend=Depends(get_state_backend))

Get focused status information for a job.

Lightweight endpoint for polling job progress.

Parameters:

Name	Type	Description	Default
job_id	str	Unique job identifier	required
backend	StateBackend	State backend (injected)	Depends(get_state_backend)

Returns:

Type	Description
JobStatusResponse	Job status summary

Raises:

Type	Description
HTTPException	404 if job not found

Source code in src/marianne/dashboard/routes/__init__.py

@router.get("/jobs/{job_id}/status", response_model=JobStatusResponse)
async def get_job_status(
    job_id: str,
    backend: StateBackend = Depends(get_state_backend),
) -> JobStatusResponse:
    """Get focused status information for a job.

    Lightweight endpoint for polling job progress.

    Args:
        job_id: Unique job identifier
        backend: State backend (injected)

    Returns:
        Job status summary

    Raises:
        HTTPException: 404 if job not found
    """
    state = await get_job_or_404(backend, job_id)
    return JobStatusResponse.from_checkpoint(state)