diff --git a/pillar_tool/routers/top.py b/pillar_tool/routers/top.py index 4fbf056..6c6efae 100644 --- a/pillar_tool/routers/top.py +++ b/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: