Reload metadata from the function directory.
(self)
| 41 | self._module_index: Dict[str, List[str]] = {} |
| 42 | |
| 43 | def refresh(self) -> None: |
| 44 | """Reload metadata from the function directory.""" |
| 45 | self._metadata.clear() |
| 46 | self._module_index = {} |
| 47 | self._load_error = None |
| 48 | manager = get_function_manager(self._functions_dir) |
| 49 | try: |
| 50 | manager.load_functions() |
| 51 | except Exception as exc: # pragma: no cover - propagated via catalog usage |
| 52 | self._loaded = True |
| 53 | self._load_error = exc |
| 54 | return |
| 55 | |
| 56 | module_index: Dict[str, List[str]] = {} |
| 57 | for name, fn in manager.list_functions().items(): |
| 58 | try: |
| 59 | metadata = _build_function_metadata(name, fn, self._functions_dir) |
| 60 | self._metadata[name] = metadata |
| 61 | module_bucket = module_index.setdefault(metadata.module_name, []) |
| 62 | module_bucket.append(name) |
| 63 | except Exception as exc: # pragma: no cover - guarded to avoid cascading failures |
| 64 | print(f"[FunctionCatalog] Failed to load metadata for {name}: {exc}") |
| 65 | for module_name, names in module_index.items(): |
| 66 | names.sort() |
| 67 | self._module_index = module_index |
| 68 | self._loaded = True |
| 69 | |
| 70 | def _ensure_loaded(self) -> None: |
| 71 | if not self._loaded: |
no test coverage detected