Peyton DeNiro
29defc5ef1
Critical fixes: - node_runtime.py: change outer handler in _load_module() from except Exception to except BaseException with sys.modules.pop() cleanup, preventing non-Exception BaseExceptions (SystemExit, KeyboardInterrupt) from violating the no-raise contract - app.py: remove incorrect reassignment of base_node_class_mappings after custom node import; keep it as pre-custom-node baseline so WorkflowPlanner correctly treats extras/custom nodes as NODE_CLASS_MAPPINGS dict lookups instead of direct imports - Regenerate stale test fixtures (text-to-image.py, upscale-model-loader.py) from current source: updated _load_module() with exec failure cleanup pattern, 3-marker structural verification, and try/finally sys.path restoration in import_custom_nodes() Suggestions addressed: - node_runtime.py: _is_comfyui_directory() now checks 3 structural markers (nodes.py, main.py, comfy/) instead of just nodes.py - generator/render.py: added None guard for getattr on generated_helpers.__all__ items with log warning for missing helpers - test_import_path_resolution.py: fixed test_sys_path_restored_on_crash to target the server module specifically after sys.path manipulation, so the try/finally restoration path is actually exercised Nits addressed: - .gitignore: added missing trailing newline New test coverage (tests/test_app_base_mappings.py): - base_node_class_mappings unchanged after custom node import - deep copy independence of base_node_class_mappings - render output contains hardened _load_module with except BaseException pattern
86 lines
3.3 KiB
Python
86 lines
3.3 KiB
Python
import copy
|
|
import sys
|
|
|
|
from typing import TextIO
|
|
|
|
from .generator.planner import WorkflowPlanner
|
|
from .generator.render import WorkflowRenderer
|
|
from .io import write_python_output
|
|
from .load_order import LoadOrderDeterminer
|
|
from .workflow_loader import load_frontend_workflow_data, load_workflow_data
|
|
|
|
|
|
class ExportApplication:
|
|
"""High-level exporter orchestration."""
|
|
|
|
def __init__(
|
|
self,
|
|
workflow: str = "",
|
|
frontend_workflow: str | dict | None = None,
|
|
input_file: str = "",
|
|
output_file: str | TextIO = "",
|
|
queue_size: int = 1,
|
|
node_class_mappings: dict | None = None,
|
|
needs_init_custom_nodes: bool = False,
|
|
node_mapping_loader=None,
|
|
custom_node_importer=None,
|
|
):
|
|
if input_file and workflow:
|
|
raise ValueError("Can't provide both input_file and workflow")
|
|
if not input_file and not workflow:
|
|
raise ValueError("Needs input_file or workflow")
|
|
if not output_file:
|
|
raise ValueError("Needs output_file")
|
|
|
|
self.workflow = workflow
|
|
self.frontend_workflow = frontend_workflow
|
|
self.input_file = input_file
|
|
self.output_file = output_file
|
|
self.queue_size = queue_size
|
|
self.node_mapping_loader = node_mapping_loader
|
|
self.custom_node_importer = custom_node_importer
|
|
self.node_class_mappings = (
|
|
node_class_mappings
|
|
if node_class_mappings is not None
|
|
else self.node_mapping_loader()
|
|
)
|
|
self.needs_init_custom_nodes = needs_init_custom_nodes
|
|
self.base_node_class_mappings = copy.deepcopy(self.node_class_mappings)
|
|
|
|
def execute(self) -> None:
|
|
data = load_workflow_data(self.workflow, self.input_file)
|
|
metadata_workflow_data = load_frontend_workflow_data(self.frontend_workflow)
|
|
|
|
missing_node_types = {
|
|
node_data["class_type"]
|
|
for node_data in data.values()
|
|
if node_data["class_type"] not in self.node_class_mappings
|
|
}
|
|
if self.needs_init_custom_nodes or missing_node_types:
|
|
self.custom_node_importer()
|
|
# Re-read from the cached "nodes" module in sys.modules after
|
|
# import_custom_nodes() populates it with extras (comfy_extras,
|
|
# custom node directories, etc.). The original dict is a stale copy.
|
|
nodes_mod = sys.modules.get("nodes")
|
|
if nodes_mod is not None:
|
|
fresh_mappings = getattr(nodes_mod, "NODE_CLASS_MAPPINGS", {})
|
|
self.node_class_mappings = fresh_mappings
|
|
# Leave self.base_node_class_mappings unchanged — it represents the
|
|
# pre-custom-node baseline used by WorkflowPlanner to decide whether
|
|
# a node gets a direct import or a NODE_CLASS_MAPPINGS dict lookup.
|
|
|
|
load_order = LoadOrderDeterminer(
|
|
data, self.node_class_mappings
|
|
).determine_load_order()
|
|
plan = WorkflowPlanner(
|
|
self.node_class_mappings, self.base_node_class_mappings
|
|
).build_plan(
|
|
load_order,
|
|
data,
|
|
metadata_workflow_data,
|
|
queue_size=self.queue_size,
|
|
)
|
|
generated_code = WorkflowRenderer().render(plan)
|
|
write_python_output(self.output_file, generated_code)
|
|
print(f"Code successfully generated and written to {self.output_file}")
|