artifacts

artifacts

Artifacts API endpoints for workspace file access.

Classes

FileInfo

Bases: BaseModel

Information about a workspace file.

ArtifactListResponse

Bases: BaseModel

Response for listing workspace artifacts.

Functions

list_artifacts async

list_artifacts(job_id, recursive=True, include_hidden=False, file_pattern=None, backend=Depends(get_state_backend))

List files in job workspace.

Parameters:

Name	Type	Description	Default
job_id	str	Unique job identifier	required
recursive	bool	Include files in subdirectories	True
include_hidden	bool	Include hidden files (starting with .)	False
file_pattern	str | None	Glob pattern to filter files (e.g. ".md", "sheet")	None
backend	StateBackend	State backend (injected)	Depends(get_state_backend)

Returns:

Type	Description
ArtifactListResponse	List of workspace files and directories

Raises:

Type	Description
HTTPException	404 if job not found, 403 if workspace not accessible

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

@router.get("/{job_id}/artifacts", response_model=ArtifactListResponse)
async def list_artifacts(
    job_id: str,
    recursive: bool = True,
    include_hidden: bool = False,
    file_pattern: str | None = None,
    backend: StateBackend = Depends(get_state_backend),
) -> ArtifactListResponse:
    """List files in job workspace.

    Args:
        job_id: Unique job identifier
        recursive: Include files in subdirectories
        include_hidden: Include hidden files (starting with .)
        file_pattern: Glob pattern to filter files (e.g. "*.md", "sheet*")
        backend: State backend (injected)

    Returns:
        List of workspace files and directories

    Raises:
        HTTPException: 404 if job not found, 403 if workspace not accessible
    """
    # Load job state to get workspace
    state = await backend.load(job_id)
    if state is None:
        raise HTTPException(status_code=404, detail=f"Score not found: {job_id}")

    workspace = resolve_job_workspace(state, job_id)

    if not workspace.exists():
        raise HTTPException(
            status_code=404,
            detail=f"Workspace directory not found: {workspace}"
        )

    if not workspace.is_dir():
        raise HTTPException(
            status_code=403,
            detail=f"Workspace path is not a directory: {workspace}"
        )

    try:
        files: list[FileInfo] = []

        if recursive:
            # Use rglob for recursive listing
            pattern = file_pattern or "*"
            glob_pattern = f"**/{pattern}" if not pattern.startswith("**/") else pattern

            for item in workspace.glob(glob_pattern):
                # Skip hidden files/dirs unless requested
                workspace_len = len(workspace.parts)
                relative_parts = item.parts[workspace_len:]
                if not include_hidden and any(part.startswith('.') for part in relative_parts):
                    continue

                try:
                    files.append(_get_file_info(item, workspace))
                except (OSError, PermissionError):
                    # Skip files we can't access
                    continue
        else:
            # List only direct children
            for item in workspace.iterdir():
                # Skip hidden files/dirs unless requested
                if not include_hidden and item.name.startswith('.'):
                    continue

                # Apply pattern filter if provided
                if file_pattern and not item.match(file_pattern):
                    continue

                try:
                    files.append(_get_file_info(item, workspace))
                except (OSError, PermissionError):
                    # Skip files we can't access
                    continue

        # Sort files: directories first, then by name
        files.sort(key=lambda f: (f.type == "file", f.name.lower()))

        return ArtifactListResponse(
            job_id=job_id,
            workspace=str(workspace),
            total_files=len(files),
            files=files
        )

    except PermissionError as e:
        raise HTTPException(
            status_code=403,
            detail=f"Permission denied accessing workspace: {workspace}"
        ) from e
    except OSError as e:
        raise HTTPException(
            status_code=500,
            detail=f"Error listing workspace: {e}"
        ) from e

get_artifact async

get_artifact(job_id, path, download=False, backend=Depends(get_state_backend))

Get content of a specific workspace file.

Parameters:

Name	Type	Description	Default
job_id	str	Unique job identifier	required
path	str	Relative path to file within workspace	required
download	bool	If true, force download instead of inline display	False
backend	StateBackend	State backend (injected)	Depends(get_state_backend)

Returns:

Type	Description
Response	File content with appropriate content type

Raises:

Type	Description
HTTPException	404 if job/file not found, 403 if not accessible, 400 if path invalid

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

@router.get("/{job_id}/artifacts/{path:path}")
async def get_artifact(
    job_id: str,
    path: str,
    download: bool = False,
    backend: StateBackend = Depends(get_state_backend),
) -> Response:
    """Get content of a specific workspace file.

    Args:
        job_id: Unique job identifier
        path: Relative path to file within workspace
        download: If true, force download instead of inline display
        backend: State backend (injected)

    Returns:
        File content with appropriate content type

    Raises:
        HTTPException: 404 if job/file not found, 403 if not accessible, 400 if path invalid
    """
    # Load job state to get workspace
    state = await backend.load(job_id)
    if state is None:
        raise HTTPException(status_code=404, detail=f"Score not found: {job_id}")

    workspace = resolve_job_workspace(state, job_id)

    if not workspace.exists() or not workspace.is_dir():
        raise HTTPException(
            status_code=404,
            detail=f"Workspace not accessible: {workspace}"
        )

    # Validate path safety (prevent directory traversal)
    if not _is_safe_path(path, workspace):
        raise HTTPException(
            status_code=400,
            detail=f"Invalid path: {path}"
        )

    file_path = workspace / path

    if not file_path.exists():
        raise HTTPException(
            status_code=404,
            detail=f"File not found: {path}"
        )

    if not file_path.is_file():
        raise HTTPException(
            status_code=400,
            detail=f"Path is not a file: {path}"
        )

    try:
        # Check if file is readable
        if not os.access(file_path, os.R_OK):
            raise HTTPException(
                status_code=403,
                detail=f"Permission denied: {path}"
            )

        # Get MIME type
        mime_type, _ = mimetypes.guess_type(str(file_path))

        # Determine if file should be served inline or as download
        if download:
            # Force download
            return FileResponse(
                path=file_path,
                filename=file_path.name,
                media_type=mime_type or "application/octet-stream"
            )

        # Serve inline for text files, download for binary
        if mime_type and mime_type.startswith(('text/', 'application/json', 'application/xml')):
            # Read text file content
            try:
                content = file_path.read_text(encoding='utf-8')
                return PlainTextResponse(
                    content=content,
                    media_type=mime_type or "text/plain"
                )
            except UnicodeDecodeError:
                # File claimed to be text but isn't valid UTF-8, serve as binary
                return FileResponse(
                    path=file_path,
                    filename=file_path.name,
                    media_type="application/octet-stream"
                )
        else:
            # Binary file - serve as download
            return FileResponse(
                path=file_path,
                filename=file_path.name,
                media_type=mime_type or "application/octet-stream"
            )

    except PermissionError as e:
        raise HTTPException(
            status_code=403,
            detail=f"Permission denied: {path}"
        ) from e
    except OSError as e:
        raise HTTPException(
            status_code=500,
            detail=f"Error reading file: {e}"
        ) from e