Merge branch 'main' of gitea.linvogel.ch:linus/PillarTool

pillar_tool/routers/top.py

@@ -38,8 +38,8 @@ def top_get(req: Request, host: str):
         print("[DEBUG] top_get: resolved host '{}' with id={}".format(host, target_host.id))
 
     parent_stmt = select(Host).where(Host.id == bindparam("parent_id"))
-    parents = []
-    current: Host = target_host
+    parents: list[Host] = []
+    current: Host | None = target_host
     while current is not None and current.parent_id is not None:
         parents.append(current)
         result = db.execute(parent_stmt, {'parent_id': current.parent_id}).fetchall()
@@ -64,7 +64,7 @@ def top_get(req: Request, host: str):
     for host in reversed(parents):
         env_res = db.execute(env_stmt, {'host_id': host.id}).fetchall()
         if len(env_res) == 1:
-            env: Environment = env_res[0][0]
+            env: Environment | None = env_res[0][0]
 
     if env is None:
         print("[DEBUG] top_get: WARNING - No environment assigned to host '{}' or any of its ancestors in the hierarchy".format(host))
@@ -72,7 +72,7 @@ def top_get(req: Request, host: str):
     state_stmt = (select(State)
         .join(TopFile, State.id == TopFile.state_id)
         .join(StateAssignment, State.id == StateAssignment.state_id)
-        .where(and_(StateAssignment.environment_id == env.id, TopFile.host_id == bindparam("host_id")))
+        .where(and_(StateAssignment.environment_id == env.id, TopFile.host_id == bindparam("host_id"))) # type: ignore
     )
     assigned_states = [
         [ row[0] for row in db.execute(state_stmt, {'host_id': host.id}).fetchall() ]
@@ -80,12 +80,12 @@ def top_get(req: Request, host: str):
     ]
 
     all_assigned_states = set(s for states in assigned_states for s in states)
-    env_name = env.name
+    env_name = env.name # type: ignore
 
-    print("[DEBUG] top_get: found {} assigned state(s) for host '{}' in environment '{}': {}".format(len(all_assigned_states), host, env.name, [s.name for s in all_assigned_states]))
+    print("[DEBUG] top_get: found {} assigned state(s) for host '{}' in environment '{}': {}".format(len(all_assigned_states), host, env.name, [s.name for s in all_assigned_states])) # type: ignore
 
     return JSONResponse(status_code=200, content={
-        env.name: list(map(lambda state: state.name, all_assigned_states)),
+        env.name: list(map(lambda state: state.name, all_assigned_states)), # type: ignore
     })
 
 
@@ -143,15 +143,15 @@ def top_state_assign(req: Request, host_name: str, state_name: str):
             print("[DEBUG] top_state_assign: ERROR - Host '{}' not found at path level '{}'. Expected exactly one match but got {} results.".format(host_name, path, len(host_res)))
             return JSONResponse(status_code=404, content={"error": f"Host '{host_name} not found"})
 
-        current: Host = host_res[0][0]
-        parent_id = current.id
+        current: Host | None = host_res[0][0] # NOTE: this result cannot be none, so the next line is safe
+        parent_id = current.id # type: ignore
 
     host: Host = current
-    print("[DEBUG] top_state_assign: resolved target host '{}' with id={}".format(host.name, host.id))
+    print("[DEBUG] top_state_assign: resolved target host '{}' with id={}".format(host.name, host.id)) # type: ignore
 
     parent_stmt = select(Host).where(Host.id == bindparam("parent_id"))
     parents: list[Host] = []
-    current: Host = host
+    current: Host | None = host
     while current is not None:
         parents.append(current)
         if current.parent_id is None:
@@ -162,7 +162,7 @@ def top_state_assign(req: Request, host_name: str, state_name: str):
                 print("[DEBUG] top_state_assign: ERROR - Host Hierarchy seems broken: host '{}' has parent_id '{}' which does not exist in database. This indicates a foreign key constraint violation or orphaned record.".format(current.name, current.parent_id))
                 return JSONResponse(status_code=500, content={"error": f"Host Hierarchy seems broken: parent_id '{current.parent_id}' does not exist"})
             # Note: more than one result is impossible, since the id is a primary key
-            current: Host = parent[0][0]
+            current: Host | None = parent[0][0]
 
     print("[DEBUG] top_state_assign: resolved parent hierarchy with {} hosts".format(len(parents)))
 
@@ -172,7 +172,7 @@ def top_state_assign(req: Request, host_name: str, state_name: str):
     for current_host in parents:
         env_res = db.execute(env_assign_stmt, {'host_id': current_host.id}).fetchall()
         if len(env_res) == 1:
-            env_assign: EnvironmentAssignment = env_res[0][0]
+            env_assign: EnvironmentAssignment | None = env_res[0][0]
             break
 
     if env_assign is None:
@@ -197,7 +197,7 @@ def top_state_assign(req: Request, host_name: str, state_name: str):
     state: State = state_res[0][0]
 
     # insert the relation into the database
-    db.execute(insert(TopFile).on_conflict_do_nothing('pillar_tool_top_file_unique_state_host').values(state_id=state.id, host_id=host.id))
+    db.execute(insert(TopFile).on_conflict_do_nothing('pillar_tool_top_file_unique_state_host').values(state_id=state.id, host_id=host.id)) # type: ignore
     print("[DEBUG] top_state_assign: successfully assigned state '{}' to host '{}' in environment '{}'".format(state_name, host_name, env.name))
 
     return JSONResponse(status_code=200, content={})
@@ -221,7 +221,7 @@ def top_state_unassign(req: Request, host_name: str, state_name: str):
 
     parent_stmt = select(Host).where(Host.id == bindparam("parent_id"))
     parents: list[Host] = []
-    current: Host = host
+    current: Host | None = host
     while current is not None:
         parents.append(current)
         if current.parent_id is None:
@@ -232,7 +232,7 @@ def top_state_unassign(req: Request, host_name: str, state_name: str):
                 print("[DEBUG] top_state_unassign: ERROR - Host Hierarchy seems broken: host '{}' has parent_id '{}' which does not exist in database. This indicates a foreign key constraint violation or orphaned record.".format(current.name, current.parent_id))
                 return JSONResponse(status_code=500, content={"error": f"Host Hierarchy seems broken: parent_id '{current.parent_id}' does not exist"})
             # Note: more than one result is impossible, since the id is a primary key
-            current: Host = parent[0][0]
+            current: Host | None = parent[0][0]
 
     print("[DEBUG] top_state_unassign: resolved parent hierarchy with {} hosts".format(len(parents)))
 
@@ -242,7 +242,7 @@ def top_state_unassign(req: Request, host_name: str, state_name: str):
     for current_host in parents:
         env_res = db.execute(env_assign_stmt, {'host_id': current_host.id}).fetchall()
         if len(env_res) == 1:
-            env_assign: EnvironmentAssignment = env_res[0][0]
+            env_assign: EnvironmentAssignment | None = env_res[0][0]
             break
 
     if env_assign is None: