Coverage for core / optional_import.py: 0.0%

1""" 

2core.optional_import — logged graceful degradation for optional dependencies. 

3 

4WHY THIS EXISTS 

5─────────────── 

6Nunba had at least 6 `try: import X; except ImportError: pass` blocks 

7sprinkled across main.py and adjacent modules. Each one silently swallowed 

8the failure with no log line, no metric, no operator-visible signal. When 

9a feature mysteriously stopped working in a frozen build (because a wheel 

10got pruned, a sys.path entry shifted, or a transitive dep flipped backend), 

11there was NO way to find out from the running process — operators had to 

12add print statements and rebuild. 

13 

14This module replaces that pattern with a single helper that: 

15 1. Attempts the import. 

16 2. On failure, logs ONE INFO-level line with the human-readable reason 

17 (so it lands in the standard log, surfaces in /api/admin/logs). 

18 3. Records the degradation in a process-global registry queryable via 

19 `/api/admin/diag/degradations` (operator self-service diagnosis). 

20 4. Returns a sentinel/fallback so the call site can keep using the name 

21 without `if X is None:` peppering. 

22 5. Is idempotent — the second call for the same module is silent. 

23 

24DESIGN NOTES 

25──────────── 

26- We log at INFO, not WARNING. These are EXPECTED degradations (e.g., 

27 optional GPU libs missing on a CPU-only laptop). WARNING would cry-wolf 

28 and train operators to ignore the channel. 

29- The registry is a plain dict — no thread safety needed because all 

30 imports happen at module-load time on the main thread. 

31- We do NOT cache the imported module here; let Python's normal import 

32 machinery do that. This module ONLY tracks failures. 

33- The Flask blueprint exposes the registry as JSON; gated by 

34 `require_local_or_token` upstream because degradation lists can leak 

35 the absence of paid-tier integrations (an information-disclosure vector 

36 on multi-tenant deploys, irrelevant on flat). 

37""" 

38from __future__ import annotations 

39 

40import importlib 

41import logging 

42import time 

43from typing import Any, Dict, List, Optional 

44 

45logger = logging.getLogger(__name__) 

46 

47 

48# Registry of failed imports — process-global, single-source-of-truth for 

49# the /api/admin/diag/degradations endpoint. 

50# key: module name (e.g., "integrations.service_tools.vram_manager") 

51# value: {reason, error, ts, attempts} 

52_DEGRADED: Dict[str, Dict[str, Any]] = {} 

53 

54# Modules we already successfully imported — used to short-circuit repeat 

55# calls so the cost of `optional_import('foo', ...)` in a hot path is one 

56# dict lookup, not a fresh importlib walk. 

57_LOADED: Dict[str, Any] = {} 

58 

59 

60def optional_import( 

61 module_name: str, 

62 reason: str, 

63 fallback: Any = None, 

64) -> Any: 

65 """Import a module by name; on failure log + register and return fallback. 

66 

67 Args: 

68 module_name: Dotted import path, e.g. ``'integrations.service_tools.vram_manager'``. 

69 reason: Human-readable why-this-is-optional, used in the log line 

70 AND surfaced in `/api/admin/diag/degradations`. Examples: 

71 ``'GPU VRAM telemetry'``, ``'HF Hub model search'``, 

72 ``'WAMP ticket auth'``. Be specific — "feature unavailable" 

73 is unhelpful. 

74 fallback: Value returned when the import fails. Defaults to None. 

75 Pass a stub class or no-op module if call sites need attribute 

76 access. 

77 

78 Returns: 

79 The imported module on success, or `fallback` on ImportError / 

80 any other exception during import. 

81 

82 Idempotency: 

83 Successful imports are cached in `_LOADED` and returned directly 

84 on subsequent calls (no re-import). Failed imports increment an 

85 attempt counter in `_DEGRADED` but do NOT re-log — first failure 

86 is the signal, subsequent retries are noise. 

87 """ 

88 if module_name in _LOADED: 

89 return _LOADED[module_name] 

90 

91 if module_name in _DEGRADED: 

92 _DEGRADED[module_name]['attempts'] += 1 

93 return fallback 

94 

95 try: 

96 mod = importlib.import_module(module_name) 

97 _LOADED[module_name] = mod 

98 return mod 

99 except Exception as e: 

100 # Catch broad — `ImportError` misses things like circular-import 

101 # `AttributeError` and missing-DLL `OSError` on Windows. 

102 _DEGRADED[module_name] = { 

103 'reason': reason, 

104 'error': f"{type(e).__name__}: {e}", 

105 'ts': time.time(), 

106 'attempts': 1, 

107 } 

108 # ONE log line per module — INFO level (expected degradation, not 

109 # a panic). WARNING would cry-wolf for legitimate optional deps. 

110 logger.info( 

111 "optional_import: %s unavailable (%s) — %s: %s", 

112 module_name, reason, type(e).__name__, e, 

113 ) 

114 return fallback 

115 

116 

117def list_degradations() -> List[Dict[str, Any]]: 

118 """Return a snapshot of all registered degradations for the admin endpoint. 

119 

120 Output is a list of dicts with stable keys so the frontend can render 

121 a table without per-field probing. Sorted by first-failure timestamp 

122 so the UI naturally shows boot-time degradations first.""" 

123 out = [] 

124 for name, info in _DEGRADED.items(): 

125 out.append({ 

126 'module': name, 

127 'reason': info['reason'], 

128 'error': info['error'], 

129 'first_failed_at': info['ts'], 

130 'attempts': info['attempts'], 

131 }) 

132 out.sort(key=lambda d: d['first_failed_at']) 

133 return out 

134 

135 

136def is_available(module_name: str) -> bool: 

137 """Cheap predicate for call sites that need a boolean check before 

138 invoking a feature — avoids the fallback-sentinel dance.""" 

139 return module_name in _LOADED 

140 

141 

142def reset_for_tests() -> None: 

143 """Clear both registries. Test-only helper — do NOT call from app code.""" 

144 _DEGRADED.clear() 

145 _LOADED.clear() 

146 

147 

148__all__ = [ 

149 'optional_import', 

150 'list_degradations', 

151 'is_available', 

152 'reset_for_tests', 

153]