57 lines
1.3 KiB
Python
57 lines
1.3 KiB
Python
from typing import Type, 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
|
|
|
|
|
|
def get_hostgroup_params_from_query(req: Request):
|
|
return HostgroupParams(**dict(req.query_params))
|
|
|
|
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
|