inecs
e8b08526d1
Return native JSON with the missing id in the message, clear stale sessions on 401, and document ingress-nginx annotations so branded HTML 404/405 pages do not rewrite simulator bodies.
130 lines
3.3 KiB
Python
130 lines
3.3 KiB
Python
"""vSphere Automation API error shapes."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from typing import Any
|
|
|
|
from fastapi import HTTPException
|
|
|
|
|
|
class VsphereError(HTTPException):
|
|
"""HTTPException with a vSphere Automation-style JSON body."""
|
|
|
|
def __init__(
|
|
self,
|
|
status_code: int,
|
|
*,
|
|
error_type: str,
|
|
messages: list[dict[str, Any]] | None = None,
|
|
data: dict[str, Any] | None = None,
|
|
) -> None:
|
|
detail: dict[str, Any] = {
|
|
"error_type": error_type,
|
|
"messages": messages or [{"default_message": error_type, "id": error_type, "args": []}],
|
|
}
|
|
if data is not None:
|
|
detail["data"] = data
|
|
super().__init__(status_code=status_code, detail=detail)
|
|
|
|
|
|
def unauthenticated(message: str = "Authentication required") -> VsphereError:
|
|
return VsphereError(
|
|
401,
|
|
error_type="unauthenticated",
|
|
messages=[
|
|
{
|
|
"default_message": message,
|
|
"id": "com.vmware.vapi.endpoint.unauthenticated",
|
|
"args": [],
|
|
}
|
|
],
|
|
)
|
|
|
|
|
|
def not_found(message: str = "Not found", *, args: list[Any] | None = None) -> VsphereError:
|
|
return VsphereError(
|
|
404,
|
|
error_type="not_found",
|
|
messages=[
|
|
{
|
|
"default_message": message,
|
|
"id": "com.vmware.vapi.std.errors.not_found",
|
|
"args": list(args or []),
|
|
}
|
|
],
|
|
)
|
|
|
|
|
|
def no_such(resource: str, resource_id: str) -> VsphereError:
|
|
"""Native not-found with the missing id in the message (Ingress-friendly JSON)."""
|
|
|
|
message = f"No such {resource} ('{resource_id}')"
|
|
return VsphereError(
|
|
404,
|
|
error_type="not_found",
|
|
messages=[
|
|
{
|
|
"default_message": message,
|
|
"id": "com.vmware.vapi.std.errors.not_found",
|
|
"args": [resource_id],
|
|
}
|
|
],
|
|
data={resource: message},
|
|
)
|
|
|
|
|
|
def already_exists(message: str = "Already exists") -> VsphereError:
|
|
return VsphereError(
|
|
400,
|
|
error_type="already_exists",
|
|
messages=[
|
|
{
|
|
"default_message": message,
|
|
"id": "com.vmware.vapi.std.errors.already_exists",
|
|
"args": [],
|
|
}
|
|
],
|
|
)
|
|
|
|
|
|
def invalid_argument(message: str = "Invalid argument") -> VsphereError:
|
|
return VsphereError(
|
|
400,
|
|
error_type="invalid_argument",
|
|
messages=[
|
|
{
|
|
"default_message": message,
|
|
"id": "com.vmware.vapi.std.errors.invalid_argument",
|
|
"args": [],
|
|
}
|
|
],
|
|
)
|
|
|
|
|
|
def unauthorized(message: str = "Unauthorized") -> VsphereError:
|
|
return VsphereError(
|
|
403,
|
|
error_type="unauthorized",
|
|
messages=[
|
|
{
|
|
"default_message": message,
|
|
"id": "com.vmware.vapi.std.errors.unauthorized",
|
|
"args": [],
|
|
}
|
|
],
|
|
)
|
|
|
|
|
|
def not_implemented(message: str = "Not implemented for active contract version") -> VsphereError:
|
|
return VsphereError(
|
|
501,
|
|
error_type="error",
|
|
messages=[
|
|
{
|
|
"default_message": message,
|
|
"id": "com.vmware.vapi.std.errors.error",
|
|
"args": [],
|
|
}
|
|
],
|
|
)
|