from typing import Any from pillar_tool.db.models.pillar_data import * from sqlalchemy import select, insert, union from sqlalchemy.orm import Session def get_pillar_name_sequence(name: str) -> list[str]: return name.split(':') def generate_host_hierarchy(db: Session, labels: list[str]) -> list[Host]: path_consumed = [] out = [] last_parent_id = None for label in labels: path_consumed += label stmt = select(Host).where(Host.name == label and Host.parent_id == last_parent_id) result = list(db.execute(stmt).fetchall()) if not result: raise RuntimeError(f"No such host(-group): '{':'.join(path_consumed)}'") # NOTE: this is an assertion because the schema should enforce this assert len(result) == 1 instance = Host(result[0]) print(instance.id) out.append(instance) return out def get_values_for_host(db: Session, host: str) -> dict: labels = get_pillar_name_sequence(host) hierarchy = generate_host_hierarchy(db, labels) # TODO: generate host hierarchy # TODO: find all values assigned o this host hierarchy and sort by depth # TODO: build the pillar structure return {} def create_pillar_host(db: Session, host_id: UUID, name: str, value: Any) -> None: # TODO: generate host hierarchy # get the involved host or hostgroup res = db.execute(select(Host).where(Host.id == host_id)).fetchone() if res is None: # TODO: handle this error with a custom Exception raise RuntimeError(f"No Host or Hostgroup with id {host_id} exists!") host = res[0][0] # TODO: generate pillar path from name # TODO: find if pillar already exists # TODO: create new pillar if it doesn't exist # TODO: assign value to new or existing pillar return def create_pillar_host_group(db: Session, host_group: UUID, name: str, value: Any) -> None: pass