Class FunctionManager

utils/function_manager.py:34–119

Unified function manager for loading and managing functions across the project.

class FunctionManager:
    """Unified function manager for loading and managing functions across the project."""

    def __init__(self, functions_dir: str | Path = "functions") -> None:
        self.functions_dir = Path(functions_dir)
        self.functions: Dict[str, Callable] = {}
        self._loaded = False

    def load_functions(self) -> None:
        """Load all Python functions from functions directory."""
        if self._loaded:
            return
            
        if not self.functions_dir.exists():
            raise ValueError(f"Functions directory does not exist: {self.functions_dir}")

        for file in self.functions_dir.rglob("*.py"):
            if file.name.startswith("_") or file.name == "__init__.py":
                continue
            if "__pycache__" in file.parts:
                continue
                
            module_name = self._build_module_name(file)
            try:
                # Import module dynamically
                spec = importlib.util.spec_from_file_location(module_name, file)
                if spec is None or spec.loader is None:
                    continue
                    
                module = importlib.util.module_from_spec(spec)
                spec.loader.exec_module(module)

                current_file = file.resolve()
                # Get all functions defined in the module
                for name, obj in inspect.getmembers(module, inspect.isfunction):
                    if name.startswith("_"):
                        continue
                    # Only register functions defined in the current module/file
                    if getattr(obj, "__module__", None) != module.__name__:
                        code = getattr(obj, "__code__", None)
                        source_path = Path(code.co_filename).resolve() if code else None
                        if source_path != current_file:
                            continue
                    self.functions[name] = obj
            except Exception as e:
                print(f"Error loading module {module_name}: {e}")
        
        self._loaded = True

    def _build_module_name(self, filepath: Path) -> str:
        """Create a unique module name for a function file."""
        relative = filepath.relative_to(self.functions_dir)
        parts = "_".join(relative.with_suffix("").parts) or "module"
        unique_suffix = f"{abs(hash(filepath.as_posix())) & 0xFFFFFFFF:X}"
        return f"{_MODULE_PREFIX}.{parts}_{unique_suffix}"

    def get_function(self, name: str) -> Optional[Callable]:
        """Get a function by name."""