From 053595f5667fca696bc9578b46d3475155337dfc Mon Sep 17 00:00:00 2001 From: "copilot-swe-agent[bot]" <198982749+Copilot@users.noreply.github.com> Date: Sat, 10 Jan 2026 21:57:57 +0000 Subject: [PATCH] Refactor: consolidate utilities and fix code duplication Co-authored-by: johndoe6345789 <224850594+johndoe6345789@users.noreply.github.com> --- backend/autometabuilder/utils.py | 43 ++++++++++++++++++- .../autometabuilder/workflow/plugin_loader.py | 6 +-- .../backend_build_tool_map.py | 30 ++----------- .../backend_load_metadata.py | 41 +----------------- .../backend_load_plugins.py | 6 +-- .../backend_load_tool_policies.py | 6 +-- .../backend_load_tool_registry.py | 20 +-------- .../backend_load_tools/backend_load_tools.py | 4 +- .../web_register_blueprint.py | 11 +---- 9 files changed, 60 insertions(+), 107 deletions(-) diff --git a/backend/autometabuilder/utils.py b/backend/autometabuilder/utils.py index c2cde01..891ff78 100644 --- a/backend/autometabuilder/utils.py +++ b/backend/autometabuilder/utils.py @@ -3,6 +3,7 @@ This module provides helper functions that are used across the codebase. These are pure utility functions that don't contain business logic. """ +import importlib import json import os import yaml @@ -10,6 +11,14 @@ from pathlib import Path from typing import Any +def get_package_root() -> Path: + """Get the AutoMetabuilder package root directory. + + Returns the absolute path to the autometabuilder package root. + """ + return Path(__file__).resolve().parent + + def read_json(path: Path) -> dict[str, Any]: """Read JSON file.""" if not path.exists(): @@ -18,6 +27,20 @@ def read_json(path: Path) -> dict[str, Any]: return json.load(f) +def load_callable(path: str): + """Import and return a callable by dotted path. + + Args: + path: Dotted path to callable (e.g., 'module.submodule.function') + + Returns: + The callable object + """ + module_path, attr = path.rsplit(".", 1) + module = importlib.import_module(module_path) + return getattr(module, attr) + + def load_metadata() -> dict[str, Any]: """Load metadata.json with optional section includes. @@ -29,8 +52,8 @@ def load_metadata() -> dict[str, Any]: "workflow_plugins_path": "workflow_plugins", } - # Locate metadata.json relative to the autometabuilder package root - metadata_path = Path(__file__).resolve().parent / "metadata.json" + # Locate metadata.json in package root + metadata_path = get_package_root() / "metadata.json" metadata = read_json(metadata_path) base_dir = metadata_path.parent @@ -60,3 +83,19 @@ def load_prompt_yaml() -> dict: with open(local_path, "r", encoding="utf-8") as f: return yaml.safe_load(f) raise FileNotFoundError(f"Prompt file not found at {local_path}") + + +def load_tool_registry() -> list: + """Load tool registry entries from tool_registry.json. + + This is a utility function for loading tool registry configuration. + """ + path = get_package_root() / "tool_registry.json" + if not os.path.exists(path): + return [] + try: + with open(path, "r", encoding="utf-8") as f: + data = json.load(f) + except json.JSONDecodeError: + return [] + return data if isinstance(data, list) else [] diff --git a/backend/autometabuilder/workflow/plugin_loader.py b/backend/autometabuilder/workflow/plugin_loader.py index 71f7d8a..f0cfd0e 100644 --- a/backend/autometabuilder/workflow/plugin_loader.py +++ b/backend/autometabuilder/workflow/plugin_loader.py @@ -1,9 +1,7 @@ """Load workflow plugins by dotted path.""" -import importlib +from ..utils import load_callable def load_plugin_callable(path: str): """Load a workflow plugin callable.""" - module_path, attr = path.rsplit(".", 1) - module = importlib.import_module(module_path) - return getattr(module, attr) + return load_callable(path) diff --git a/backend/autometabuilder/workflow/plugins/backend/backend_build_tool_map/backend_build_tool_map.py b/backend/autometabuilder/workflow/plugins/backend/backend_build_tool_map/backend_build_tool_map.py index 1e5dcd8..c9e85cd 100644 --- a/backend/autometabuilder/workflow/plugins/backend/backend_build_tool_map/backend_build_tool_map.py +++ b/backend/autometabuilder/workflow/plugins/backend/backend_build_tool_map/backend_build_tool_map.py @@ -1,29 +1,5 @@ """Workflow plugin: build tool map.""" -import importlib -import json -import os -from pathlib import Path - - -def _load_callable(path: str): - """Import and return a callable.""" - module_path, attr = path.rsplit(".", 1) - module = importlib.import_module(module_path) - return getattr(module, attr) - - -def _load_tool_registry() -> list: - """Load tool registry entries.""" - # Locate tool_registry.json relative to autometabuilder package root - path = Path(__file__).resolve().parents[4] / "tool_registry.json" - if not os.path.exists(path): - return [] - try: - with open(path, "r", encoding="utf-8") as f: - data = json.load(f) - except json.JSONDecodeError: - return [] - return data if isinstance(data, list) else [] +from .....utils import load_callable, load_tool_registry def _build_tool_map(gh, registry_entries: list) -> dict: @@ -40,7 +16,7 @@ def _build_tool_map(gh, registry_entries: list) -> dict: continue if provider == "module": path = entry.get("callable") - tool_map[name] = _load_callable(path) if path else None + tool_map[name] = load_callable(path) if path else None continue tool_map[name] = None return tool_map @@ -49,7 +25,7 @@ def _build_tool_map(gh, registry_entries: list) -> dict: def run(runtime, _inputs): """Build tool registry map.""" gh = runtime.context.get("gh") - registry = _load_tool_registry() + registry = load_tool_registry() tool_map = _build_tool_map(gh, registry) # Store in both store (for workflow) and context (for other plugins) runtime.context["tool_map"] = tool_map diff --git a/backend/autometabuilder/workflow/plugins/backend/backend_load_metadata/backend_load_metadata.py b/backend/autometabuilder/workflow/plugins/backend/backend_load_metadata/backend_load_metadata.py index 690c182..5e80934 100644 --- a/backend/autometabuilder/workflow/plugins/backend/backend_load_metadata/backend_load_metadata.py +++ b/backend/autometabuilder/workflow/plugins/backend/backend_load_metadata/backend_load_metadata.py @@ -1,47 +1,10 @@ """Workflow plugin: load metadata.""" -import json -from pathlib import Path -from typing import Any - - -INCLUDED_SECTIONS = { - "settings_descriptions_path": "settings_descriptions", - "suggestions_path": "suggestions", - "workflow_plugins_path": "workflow_plugins", -} - - -def _read_json(path: Path) -> dict[str, Any]: - """Read JSON file.""" - if not path.exists(): - return {} - with path.open("r", encoding="utf-8") as f: - return json.load(f) - - -def _load_metadata() -> dict[str, Any]: - """Load metadata.json with optional section includes.""" - # Locate metadata.json relative to the autometabuilder package root - metadata_path = Path(__file__).resolve().parents[4] / "metadata.json" - metadata = _read_json(metadata_path) - base_dir = metadata_path.parent - for path_key, dest_key in INCLUDED_SECTIONS.items(): - include_path = metadata.get(path_key) - if include_path: - resolved_path = base_dir / include_path - if resolved_path.is_dir(): - merged: dict[str, Any] = {} - for file_path in sorted(resolved_path.glob("*.json")): - merged.update(_read_json(file_path)) - metadata[dest_key] = merged - else: - metadata[dest_key] = _read_json(resolved_path) - return metadata +from .....utils import load_metadata as util_load_metadata def run(runtime, _inputs): """Load metadata.json.""" - metadata = _load_metadata() + metadata = util_load_metadata() # Store in both store (for workflow) and context (for other plugins) runtime.context["metadata"] = metadata return {"result": metadata} diff --git a/backend/autometabuilder/workflow/plugins/backend/backend_load_plugins/backend_load_plugins.py b/backend/autometabuilder/workflow/plugins/backend/backend_load_plugins/backend_load_plugins.py index 80c1030..18bd7c4 100644 --- a/backend/autometabuilder/workflow/plugins/backend/backend_load_plugins/backend_load_plugins.py +++ b/backend/autometabuilder/workflow/plugins/backend/backend_load_plugins/backend_load_plugins.py @@ -3,15 +3,15 @@ import importlib import inspect import logging import os -from pathlib import Path +from .....utils import get_package_root logger = logging.getLogger("autometabuilder") def _load_plugins(tool_map: dict, tools: list) -> None: """Load plugin tools and append metadata.""" - # Locate plugins directory relative to autometabuilder package root - plugins_dir = Path(__file__).resolve().parents[4] / "plugins" + # Locate plugins directory in package root + plugins_dir = get_package_root() / "plugins" if not os.path.exists(plugins_dir): return diff --git a/backend/autometabuilder/workflow/plugins/backend/backend_load_tool_policies/backend_load_tool_policies.py b/backend/autometabuilder/workflow/plugins/backend/backend_load_tool_policies/backend_load_tool_policies.py index 61a018f..6997e38 100644 --- a/backend/autometabuilder/workflow/plugins/backend/backend_load_tool_policies/backend_load_tool_policies.py +++ b/backend/autometabuilder/workflow/plugins/backend/backend_load_tool_policies/backend_load_tool_policies.py @@ -1,13 +1,13 @@ """Workflow plugin: load tool policies.""" import json import os -from pathlib import Path +from .....utils import get_package_root def _load_tool_policies() -> dict: """Load tool policies JSON.""" - # Locate tool_policies.json relative to autometabuilder package root - path = Path(__file__).resolve().parents[4] / "tool_policies.json" + # Locate tool_policies.json in package root + path = get_package_root() / "tool_policies.json" if not os.path.exists(path): return {"modifying_tools": []} try: diff --git a/backend/autometabuilder/workflow/plugins/backend/backend_load_tool_registry/backend_load_tool_registry.py b/backend/autometabuilder/workflow/plugins/backend/backend_load_tool_registry/backend_load_tool_registry.py index 9652059..68641e1 100644 --- a/backend/autometabuilder/workflow/plugins/backend/backend_load_tool_registry/backend_load_tool_registry.py +++ b/backend/autometabuilder/workflow/plugins/backend/backend_load_tool_registry/backend_load_tool_registry.py @@ -1,26 +1,10 @@ """Workflow plugin: load tool registry.""" -import json -import os -from pathlib import Path - - -def _load_tool_registry() -> list: - """Load tool registry entries.""" - # Locate tool_registry.json relative to autometabuilder package root - path = Path(__file__).resolve().parents[4] / "tool_registry.json" - if not os.path.exists(path): - return [] - try: - with open(path, "r", encoding="utf-8") as f: - data = json.load(f) - except json.JSONDecodeError: - return [] - return data if isinstance(data, list) else [] +from .....utils import load_tool_registry as util_load_tool_registry def run(runtime, _inputs): """Load tool registry entries.""" - tool_registry = _load_tool_registry() + tool_registry = util_load_tool_registry() # Store in context for other plugins runtime.context["tool_registry"] = tool_registry return {"result": tool_registry} diff --git a/backend/autometabuilder/workflow/plugins/backend/backend_load_tools/backend_load_tools.py b/backend/autometabuilder/workflow/plugins/backend/backend_load_tools/backend_load_tools.py index 0f98365..57259d9 100644 --- a/backend/autometabuilder/workflow/plugins/backend/backend_load_tools/backend_load_tools.py +++ b/backend/autometabuilder/workflow/plugins/backend/backend_load_tools/backend_load_tools.py @@ -1,13 +1,13 @@ """Workflow plugin: load tools.""" import json import os -from pathlib import Path +from .....utils import get_package_root def _load_tools(metadata: dict) -> list: """Load tool specs from metadata reference.""" # Locate tools relative to autometabuilder package root - base_dir = Path(__file__).resolve().parents[4] + base_dir = get_package_root() tools_path = base_dir / metadata.get("tools_path", "tools.json") if tools_path.is_dir(): diff --git a/backend/autometabuilder/workflow/plugins/web/web_register_blueprint/web_register_blueprint.py b/backend/autometabuilder/workflow/plugins/web/web_register_blueprint/web_register_blueprint.py index 7b8e909..e3ab397 100644 --- a/backend/autometabuilder/workflow/plugins/web/web_register_blueprint/web_register_blueprint.py +++ b/backend/autometabuilder/workflow/plugins/web/web_register_blueprint/web_register_blueprint.py @@ -1,12 +1,5 @@ """Workflow plugin: register Flask blueprint.""" -import importlib - - -def _load_callable(path: str): - """Import and return a callable.""" - module_path, attr = path.rsplit(".", 1) - module = importlib.import_module(module_path) - return getattr(module, attr) +from .....utils import load_callable def run(runtime, inputs): @@ -34,7 +27,7 @@ def run(runtime, inputs): return {"error": "blueprint or blueprint_path is required"} try: - blueprint = _load_callable(blueprint_path) + blueprint = load_callable(blueprint_path) except Exception as e: return {"error": f"Failed to load blueprint: {str(e)}"}