From 32b685d2ae0d32092aa0bb6fe7054321b4929fe0 Mon Sep 17 00:00:00 2001 From: Matt Miller Date: Wed, 16 Sep 2026 05:44:30 +0000 Subject: [PATCH] refactor(cql): drop the now-dead normalize() family from loader.py The load_graph removal left loader.py's normalize() and its helpers (_looks_like_object_info, _looks_like_api_workflow, _from_object_info, _normalize_input, _from_api_workflow) reachable only from their own tests. No production code calls normalize: the engine builds its Graph from raw object_info directly, resilient_load_object_info is the only symbol the rest of the package imports from this module, and normalize was never in cql.__all__. Delete the normalize family and the now-unused CQLRuntimeError import, rewrite the module docstring to describe only the resilient object_info cache/fetch wrapper that remains, and delete tests/comfy_cli/cql/test_loader.py (every test in it exercised normalize). Co-Authored-By: Claude Opus 4.8 --- comfy_cli/cql/loader.py | 164 ++--------------------------- tests/comfy_cli/cql/test_loader.py | 99 ----------------- 2 files changed, 7 insertions(+), 256 deletions(-) delete mode 100644 tests/comfy_cli/cql/test_loader.py diff --git a/comfy_cli/cql/loader.py b/comfy_cli/cql/loader.py index 48f72399f..95079bab5 100644 --- a/comfy_cli/cql/loader.py +++ b/comfy_cli/cql/loader.py @@ -1,17 +1,11 @@ -"""Shape and load CQL ``object_info`` graphs. +"""Load CQL ``object_info`` graphs resiliently. -This module contains two things: - -- ``normalize`` — turn any supported input (a raw ``object_info`` dump, an - API-format workflow, or an already-shaped CQL graph) into the uniform - ``{"nodes": [...], "inputs": [...], "categories": [...]}`` dict the engine - runs on. It is intentionally permissive: anything dict-shaped that looks - like one of those formats is accepted. -- ``resilient_load_object_info`` — a cache + refresh-retry + stale-fallback - wrapper over the engine's loaders (``comfy_cli.cql.engine._load_from_file`` - / ``_load_from_target``). It auto-caches every successful fetch per host, - retries once after a token refresh on failure, and falls back to the cached - dump (with a stderr warning) when the retry still fails. +This module is the resilient object_info cache/fetch wrapper. +``resilient_load_object_info`` wraps the engine's loaders +(``comfy_cli.cql.engine._load_from_file`` / ``_load_from_target``) with a +cache-first TTL gate, auto-caches every successful fetch per host, retries +once after a token refresh on failure, and falls back to the cached dump +(with a stderr warning) when the retry still fails. The live network fetch and its security guards (loopback check, no-redirect opener, byte cap, cloud HTTPS+auth) live in ``comfy_cli.cql.engine`` — this @@ -28,152 +22,8 @@ from pathlib import Path from typing import Any -from comfy_cli.cql.errors import CQLRuntimeError from comfy_cli.file_utils import atomic_write_text, cache_dir -# ---- normalization -------------------------------------------------------- - - -def normalize(data: Any) -> dict[str, Any]: - """Turn any supported input into ``{nodes, inputs, categories}``.""" - if not isinstance(data, dict): - raise CQLRuntimeError("expected a JSON object at the top level") - - # Already CQL-shaped — trust it. - if any(isinstance(data.get(k), list) for k in ("nodes", "inputs", "categories")): - graph: dict[str, Any] = { - "nodes": list(data.get("nodes") or []), - "inputs": list(data.get("inputs") or []), - "categories": list(data.get("categories") or []), - } - return graph - - if _looks_like_object_info(data): - return _from_object_info(data) - if _looks_like_api_workflow(data): - return _from_api_workflow(data) - - raise CQLRuntimeError( - "unrecognized graph shape", - details={"keys_sample": sorted(list(data.keys()))[:10]}, - ) - - -def _looks_like_object_info(data: dict[str, Any]) -> bool: - # /object_info maps "ClassName" -> { "input": {...}, "category": "...", - # "display_name": "...", "description": "...", "output": [...], ... } - if not data: - return False - return any(isinstance(v, dict) and ("input" in v or "category" in v) for v in data.values()) - - -def _looks_like_api_workflow(data: dict[str, Any]) -> bool: - if not data: - return False - return any(isinstance(v, dict) and "class_type" in v for v in data.values()) - - -def _from_object_info(data: dict[str, Any]) -> dict[str, Any]: - nodes: list[dict[str, Any]] = [] - inputs: list[dict[str, Any]] = [] - categories: dict[str, int] = {} - - for class_name, raw in data.items(): - if not isinstance(raw, dict): - continue - category = raw.get("category") - node = { - "name": class_name, - "display_name": raw.get("display_name") or class_name, - "category": category, - "description": raw.get("description"), - "output_node": bool(raw.get("output_node", False)), - "output_types": list(raw.get("output") or []), - } - nodes.append(node) - if category: - categories[category] = categories.get(category, 0) + 1 - - sections = raw.get("input") or {} - if isinstance(sections, dict): - for section, body in sections.items(): # "required" / "optional" / "hidden" - if not isinstance(body, dict): - continue - for input_name, spec in body.items(): - inputs.append(_normalize_input(class_name, section, input_name, spec)) - - return { - "nodes": nodes, - "inputs": inputs, - "categories": [{"name": k, "node_count": v} for k, v in sorted(categories.items())], - } - - -def _normalize_input(class_name: str, section: str, name: str, spec: Any) -> dict[str, Any]: - type_name: Any = None - options: dict[str, Any] = {} - choices: list[Any] = [] - if isinstance(spec, list) and spec: - type_name = spec[0] - if isinstance(type_name, list): - choices = list(type_name) - type_name = "ENUM" - if len(spec) > 1 and isinstance(spec[1], dict): - options = dict(spec[1]) - elif isinstance(spec, str): - type_name = spec - return { - "node": class_name, - "section": section, - "name": name, - "type": type_name, - "choices": choices, - "options": options, - } - - -def _from_api_workflow(data: dict[str, Any]) -> dict[str, Any]: - nodes: list[dict[str, Any]] = [] - inputs: list[dict[str, Any]] = [] - node_ids = {str(k) for k in data} - for nid, node in data.items(): - if not isinstance(node, dict): - continue - class_type = node.get("class_type") - title = (node.get("_meta") or {}).get("title") if isinstance(node.get("_meta"), dict) else None - nodes.append( - { - "id": nid, - "name": class_type or "?", - "class_type": class_type, - "title": title, - "category": None, - } - ) - raw_inputs = node.get("inputs") or {} - if isinstance(raw_inputs, dict): - for in_name, value in raw_inputs.items(): - ref = ( - isinstance(value, list) - and len(value) == 2 - and isinstance(value[1], int) - and not isinstance(value[1], bool) - and str(value[0]) in node_ids - ) - inputs.append( - { - "node_id": nid, - "node": class_type, - "name": in_name, - "value": None if ref else value, - "ref_node": value[0] if ref else None, - "ref_slot": value[1] if ref else None, - "is_reference": ref, - } - ) - return {"nodes": nodes, "inputs": inputs, "categories": []} - - # --------------------------------------------------------------------------- # Resilient object_info loading (cache + refresh-retry + stale fallback) # --------------------------------------------------------------------------- diff --git a/tests/comfy_cli/cql/test_loader.py b/tests/comfy_cli/cql/test_loader.py deleted file mode 100644 index 543a8ef73..000000000 --- a/tests/comfy_cli/cql/test_loader.py +++ /dev/null @@ -1,99 +0,0 @@ -"""Loader tests: object_info, API workflow, and pre-shaped graph inputs.""" - -from __future__ import annotations - -import pytest - -from comfy_cli.cql.errors import CQLRuntimeError -from comfy_cli.cql.loader import normalize - -OBJECT_INFO = { - "KSampler": { - "input": { - "required": { - "seed": ["INT", {"default": 0}], - "model": ["MODEL"], - "scheduler": [["normal", "karras"]], - }, - "optional": { - "denoise": ["FLOAT", {"default": 1.0}], - }, - }, - "output": ["LATENT"], - "category": "sampling", - "display_name": "K Sampler", - "description": "samples", - }, - "CheckpointLoaderSimple": { - "input": {"required": {"ckpt_name": ["STRING"]}}, - "output": ["MODEL", "CLIP", "VAE"], - "category": "loaders", - }, -} - - -API_WORKFLOW = { - "3": { - "class_type": "KSampler", - "inputs": {"seed": 42, "model": ["4", 0]}, - "_meta": {"title": "Sampler"}, - }, - "4": { - "class_type": "CheckpointLoaderSimple", - "inputs": {"ckpt_name": "sd_xl_base.safetensors"}, - }, -} - - -def test_normalize_object_info_extracts_nodes_and_inputs(): - g = normalize(OBJECT_INFO) - names = {n["name"] for n in g["nodes"]} - assert names == {"KSampler", "CheckpointLoaderSimple"} - ks = next(n for n in g["nodes"] if n["name"] == "KSampler") - assert ks["category"] == "sampling" - assert ks["display_name"] == "K Sampler" - assert ks["output_types"] == ["LATENT"] - # Inputs were flattened with section labels. - seed = next(i for i in g["inputs"] if i["node"] == "KSampler" and i["name"] == "seed") - assert seed["type"] == "INT" - assert seed["section"] == "required" - assert seed["options"]["default"] == 0 - # Choices captured for combo inputs. - sch = next(i for i in g["inputs"] if i["name"] == "scheduler") - assert sch["type"] == "ENUM" - assert sch["choices"] == ["normal", "karras"] - - -def test_normalize_object_info_aggregates_categories(): - g = normalize(OBJECT_INFO) - by_name = {c["name"]: c["node_count"] for c in g["categories"]} - assert by_name == {"sampling": 1, "loaders": 1} - - -def test_normalize_api_workflow_marks_references(): - g = normalize(API_WORKFLOW) - nodes_by_id = {n["id"]: n for n in g["nodes"]} - assert nodes_by_id["3"]["class_type"] == "KSampler" - assert nodes_by_id["3"]["title"] == "Sampler" - seed = next(i for i in g["inputs"] if i["node_id"] == "3" and i["name"] == "seed") - assert seed["is_reference"] is False - assert seed["value"] == 42 - model = next(i for i in g["inputs"] if i["node_id"] == "3" and i["name"] == "model") - assert model["is_reference"] is True - assert model["ref_node"] == "4" - assert model["ref_slot"] == 0 - - -def test_normalize_preshaped_graph_pass_through(): - pre = { - "nodes": [{"name": "Foo"}], - "inputs": [], - "categories": [{"name": "x", "node_count": 1}], - } - g = normalize(pre) - assert g["nodes"][0]["name"] == "Foo" - - -def test_normalize_rejects_garbage(): - with pytest.raises(CQLRuntimeError): - normalize({"foo": 1, "bar": "baz"})