65 lines
1.6 KiB
Python
65 lines
1.6 KiB
Python
from typing import Callable
|
|
|
|
from pydantic import BaseModel
|
|
from starlette.requests import Request
|
|
from starlette.responses import JSONResponse
|
|
|
|
|
|
# Error returns
|
|
class ErrorDetails(BaseModel):
|
|
status: str
|
|
message: str
|
|
|
|
class ErrorReturn(BaseModel):
|
|
status_code: int
|
|
content: ErrorDetails
|
|
|
|
def response(self):
|
|
return JSONResponse(self.model_dump())
|
|
|
|
class HealthCheckError(ErrorReturn):
|
|
def __init__(self, code: int, reason: str):
|
|
super().__init__(**{
|
|
'status_code': code,
|
|
'content': ErrorDetails(
|
|
status="error",
|
|
message=f"Host is not healthy: {reason}"
|
|
)
|
|
})
|
|
|
|
class HealthCheckSuccess(ErrorReturn):
|
|
def __init__(self):
|
|
super().__init__(**{
|
|
'status_code': 200,
|
|
'content': ErrorDetails(
|
|
status='ok',
|
|
message='API is healthy'
|
|
)
|
|
})
|
|
|
|
|
|
# Host operations
|
|
class HostCreateParams(BaseModel):
|
|
parent: str | None
|
|
|
|
# Hostgroup operations
|
|
class HostgroupParams(BaseModel):
|
|
path: str | None
|
|
|
|
# State operations
|
|
class StateParams(BaseModel):
|
|
pass # No parameters needed for state operations currently
|
|
|
|
# Pillar operations
|
|
class PillarParams(BaseModel):
|
|
host: str | None
|
|
hostgroup: str | None
|
|
value: str | None # value if the pillar should be set
|
|
type: str | None # type of pillar if pillar should be set
|
|
|
|
|
|
def get_model_from_query[T](model: T) -> Callable[[Request], T]:
|
|
def aux(req: Request) -> T:
|
|
return model.model_validate(dict(req.query_params or {}))
|
|
return aux
|