spark_on_k8s.api.console

Implementation reference. These symbols describe how the project works; their presence here does not make them supported application imports. Use the documented entry points and support policy.

console_capabilities() async

Expose configuration, without probing history availability or application data.

Source code in spark_on_k8s/api/console.py

@router.get("/capabilities", response_model=_ConsoleCapabilities)
async def console_capabilities() -> _ConsoleCapabilities:
    """Expose configuration, without probing history availability or application data."""
    return _ConsoleCapabilities(history_configured=bool(APIConfiguration.SPARK_ON_K8S_API_SPARK_HISTORY_HOST))

get_application_by_pod(namespace, driver_pod) async

Resolve an existing driver reference without deriving a potentially truncated app ID.

Source code in spark_on_k8s/api/console.py

@router.get("/apps/{namespace}/pods/{driver_pod}", response_model=ConsoleApplication)
async def get_application_by_pod(namespace: str, driver_pod: str) -> ConsoleApplication:
    """Resolve an existing driver reference without deriving a potentially truncated app ID."""
    namespace_scope().require(namespace)
    try:
        client = k8s.CoreV1Api(await KubernetesClientSingleton.client())
        pod = await client.read_namespaced_pod(namespace=namespace, name=driver_pod)
    except k8s.ApiException as error:
        logger.exception("Console Kubernetes driver lookup failed")
        raise _kubernetes_error(error) from error
    except Exception as error:
        logger.exception("Console Kubernetes client failed")
        raise HTTPException(
            status_code=502,
            detail={"code": "kubernetes_unavailable", "message": "Kubernetes is unavailable."},
        ) from error
    if (pod.metadata.labels or {}).get("spark-role") != "driver":
        raise HTTPException(
            status_code=404,
            detail={"code": "application_not_found", "message": "The Spark driver was not found."},
        )
    return _application_from_pod(pod, namespace)

list_applications(namespace=None, limit=50, continue_from='') async

List one Kubernetes page without requiring cluster-wide namespace access.

Source code in spark_on_k8s/api/console.py

@router.get("/apps", response_model=ConsoleApplicationsPage)
async def list_applications(
    namespace: Annotated[
        str | None, Query(min_length=1, max_length=63, pattern=r"^[a-z0-9](?:[a-z0-9-]*[a-z0-9])?$")
    ] = None,
    limit: Annotated[int, Query(ge=1, le=200)] = 50,
    continue_from: str = "",
) -> ConsoleApplicationsPage:
    """List one Kubernetes page without requiring cluster-wide namespace access."""
    if continue_from == "no-more-items":
        return ConsoleApplicationsPage(items=[])
    scope = namespace_scope()
    namespace = scope.require(namespace or scope.default)
    try:
        client = k8s.CoreV1Api(await KubernetesClientSingleton.client())
        options = {"namespace": namespace, "label_selector": "spark-role=driver", "limit": limit}
        if continue_from:
            options["_continue"] = continue_from
        pods = await client.list_namespaced_pod(**options)
    except k8s.ApiException as error:
        logger.exception("Console Kubernetes application listing failed")
        raise _kubernetes_error(error) from error
    except Exception as error:
        logger.exception("Console Kubernetes client failed")
        raise HTTPException(
            status_code=502,
            detail={"code": "kubernetes_unavailable", "message": "Kubernetes is unavailable."},
        ) from error
    return ConsoleApplicationsPage(
        items=[_application_from_pod(pod, namespace) for pod in pods.items],
        next_cursor=pods.metadata._continue or None,
    )

list_history() async

List completed history records independently of live Kubernetes applications.

Source code in spark_on_k8s/api/console.py

@router.get("/history", response_model=ConsoleHistory)
async def list_history() -> ConsoleHistory:
    """List completed history records independently of live Kubernetes applications."""
    host = APIConfiguration.SPARK_ON_K8S_API_SPARK_HISTORY_HOST
    if not host:
        raise HTTPException(
            status_code=412,
            detail={"code": "history_not_configured", "message": "Spark History Server is not configured."},
        )
    base_url = host if "://" in host else f"http://{host}"
    try:
        response = await AsyncHttpClientSingleton.client().get(
            f"{base_url.rstrip('/')}/api/v1/applications", params={"status": "completed"}
        )
        response.raise_for_status()
        applications = TypeAdapter(list[_HistoryApplication]).validate_python(response.json())
    except (httpx.HTTPError, httpx.InvalidURL, ValueError) as error:
        logger.exception("Console Spark History Server listing failed")
        raise HTTPException(
            status_code=502,
            detail={"code": "history_unavailable", "message": "Spark History Server is unavailable."},
        ) from error
    items = []
    for application in applications:
        attempt = application.attempts[0] if application.attempts else None
        items.append(
            ConsoleApplication(
                app_id=application.id,
                name=application.name or application.id,
                status=SparkAppStatus.Unknown,
                created_at=_history_start_time(attempt) if attempt else None,
                spark_version=attempt.appSparkVersion if attempt else None,
                spark_history_proxy=True,
                source="history",
            )
        )
    return ConsoleHistory(items=items)