127 lines
3.7 KiB
Python
127 lines
3.7 KiB
Python
import uuid
|
|
|
|
from sqlalchemy import select, insert, bindparam
|
|
from sqlalchemy.orm import Session
|
|
from starlette.exceptions import HTTPException
|
|
from starlette.requests import Request
|
|
from fastapi import APIRouter
|
|
from starlette.responses import JSONResponse
|
|
|
|
from pillar_tool.db import Host
|
|
from pillar_tool.db.queries.host_queries import create_host
|
|
from pillar_tool.schemas import HostCreateParams
|
|
from pillar_tool.util.validation import validate_fqdn, split_and_validate_path
|
|
|
|
router = APIRouter(
|
|
prefix="/host",
|
|
tags=["Host"],
|
|
)
|
|
|
|
|
|
|
|
@router.get("")
|
|
def hosts_get(req: Request):
|
|
db: Session = req.state.db
|
|
|
|
result = db.execute(select(Host)).fetchall()
|
|
hosts: list[Host] = list(map(lambda x: x[0], result))
|
|
|
|
return JSONResponse(status_code=200, content=list(map(lambda x: x.name, hosts)))
|
|
|
|
|
|
|
|
@router.get("/{fqdn}")
|
|
def host_get(req: Request, fqdn: str):
|
|
db: Session = req.state.db
|
|
|
|
if not validate_fqdn(fqdn):
|
|
raise HTTPException(status_code=400, detail="Provided host is not an FQDN")
|
|
|
|
host_stmt = select(Host).where(Host.name == fqdn)
|
|
result = db.execute(host_stmt).fetchall()
|
|
|
|
if len(result) != 1:
|
|
raise HTTPException(status_code=404, detail=f"No such host found (length of result was {len(result)})")
|
|
|
|
host: Host = result[0][0]
|
|
|
|
last_parent = host
|
|
path = []
|
|
parent_stmt = select(Host).where(Host.id == bindparam('parent_id'))
|
|
while host.parent_id is not None:
|
|
result = db.execute(parent_stmt, { 'parent_id': last_parent.parent_id }).fetchall()
|
|
|
|
# Note: this assertion should be enforced by the database
|
|
assert len(result) == 1
|
|
|
|
parent = result[0][0]
|
|
path.append(parent)
|
|
last_parent = parent
|
|
|
|
path.reverse()
|
|
return JSONResponse(status_code=200, content={
|
|
"host": host.name,
|
|
"path": '/'.join(map(lambda x: x.name, path))
|
|
})
|
|
|
|
|
|
@router.post("/{fqdn}")
|
|
async def host_add(request: Request, fqdn: str, params: HostCreateParams):
|
|
db: Session = request.state.db
|
|
|
|
if not validate_fqdn(fqdn):
|
|
raise HTTPException(status_code=400, detail="Provided host is not an FQDN")
|
|
|
|
if params.parent is not None:
|
|
parent_labels = split_and_validate_path(params.parent)
|
|
if parent_labels is None:
|
|
raise HTTPException(status_code=400, detail="Provided parent is not a valid path")
|
|
else:
|
|
parent_labels = []
|
|
|
|
parent_id = None
|
|
stmt_select_respecting_parent = select(Host).where(Host.name == bindparam("label") and Host.parent_id == bindparam("parent_id"))
|
|
for label in parent_labels:
|
|
result = db.execute(stmt_select_respecting_parent, {
|
|
'label': label,
|
|
'parent_id': parent_id
|
|
}).fetchall()
|
|
|
|
if len(result) == 0:
|
|
raise HTTPException(status_code=400, detail="Parent does not exist")
|
|
|
|
# Note: this should be enforced by the database
|
|
assert len(result) == 1
|
|
|
|
parent_id = result[0][0].parent_id
|
|
|
|
new_host = Host(
|
|
id=uuid.uuid4(),
|
|
name=fqdn,
|
|
parent_id=parent_id,
|
|
is_hostgroup=False
|
|
)
|
|
stmt_create_host_with_parent = insert(Host).values(id=new_host.id, name=new_host.name, parent_id=new_host.parent_id, is_hostgroup=new_host.is_hostgroup)
|
|
db.execute(stmt_create_host_with_parent).fetchall()
|
|
|
|
# Prepare response with creation details
|
|
output = {
|
|
"message": "Host created",
|
|
"host": new_host, # return the final host in the hierarchy
|
|
}
|
|
|
|
# include the full path to the new host if it exists
|
|
if params.parent is not None:
|
|
output.update({
|
|
"path": params.parent
|
|
})
|
|
|
|
return JSONResponse(output)
|
|
|
|
|
|
@router.delete("/{fqdn}")
|
|
async def host_delete(request: Request, fqdn: str):
|
|
pass
|
|
|
|
|