Coverage for .venv/lib/python3.13/site-packages/litellm/proxy/_lazy_openapi_snapshot.py: 23%

1""" 

2Per-feature OpenAPI snapshot for lazy-loaded routers. 

3 

4The committed JSON is generated by `python -m litellm.proxy._lazy_openapi_snapshot` 

5and consumed at runtime so /openapi.json can show full route info for unloaded 

6features without importing them. check-ui-api-types.yml (mirrored locally by 

7`make check`) regenerates this file and fails when the committed copy differs, 

8then rebuilds schema.d.ts from app.openapi() with the snapshot injected. After 

9changing any lazily loaded route or this generator, rerun the module and commit 

10the JSON, then run `npm run gen:api` in ui/litellm-dashboard and commit schema.d.ts. 

11""" 

12 

13import json 

14import re 

15import sys 

16from collections.abc import Callable, Mapping 

17from dataclasses import dataclass 

18from pathlib import Path 

19from typing import TYPE_CHECKING, Final 

20 

21from typing_extensions import ReadOnly, TypedDict 

22 

23if TYPE_CHECKING: 23 ↛ 24line 23 didn't jump to line 24 because the condition on line 23 was never true

24 from fastapi import FastAPI 

25 

26 from litellm.proxy._lazy_features import LazyFeature 

27 

28SNAPSHOT_FILE: Final = Path(__file__).parent / "_lazy_openapi_snapshot.json" 

29HTTP_METHOD_SUFFIXES: Final = { 

30 "delete", 

31 "get", 

32 "head", 

33 "options", 

34 "patch", 

35 "post", 

36 "put", 

37 "trace", 

38} 

39 

40 

41def _stabilize_multi_method_route_ids(routes) -> None: 

42 """FastAPI derives route IDs from a set of methods; make snapshots stable.""" 

43 

44 for route in routes: 

45 methods = sorted(getattr(route, "methods", None) or []) 

46 if len(methods) <= 1 or not getattr(route, "path_format", None): 

47 continue 

48 

49 operation_id = f"{route.name}{route.path_format}" 

50 operation_id = re.sub(r"\W", "_", operation_id) 

51 route.unique_id = f"{operation_id}_{methods[0].lower()}" 

52 

53 

54def load_snapshot() -> dict[str, dict] | None: 

55 if not SNAPSHOT_FILE.exists(): 55 ↛ 56line 55 didn't jump to line 56 because the condition on line 55 was never true

56 return None 

57 try: 

58 with SNAPSHOT_FILE.open() as f: 

59 return json.load(f) 

60 except (json.JSONDecodeError, OSError): 

61 return None 

62 

63 

64def _normalize_operation_ids(paths: dict[str, dict]) -> None: 

65 """Make FastAPI-generated operation IDs stable for multi-method routes. 

66 

67 FastAPI derives the default operation ID suffix from the first item in the 

68 route's methods set. For routes registered with several HTTP methods, that 

69 set iteration order can vary between processes, which makes the snapshot 

70 drift even when no routes changed. 

71 """ 

72 for path_ops in paths.values(): 

73 if not isinstance(path_ops, dict): 

74 continue 

75 

76 methods = {method for method in path_ops if method in HTTP_METHOD_SUFFIXES} 

77 if not methods: 

78 continue 

79 

80 for method, operation in path_ops.items(): 

81 if method not in HTTP_METHOD_SUFFIXES or not isinstance(operation, dict): 

82 continue 

83 

84 operation_id = operation.get("operationId") 

85 if not isinstance(operation_id, str): 

86 continue 

87 

88 for suffix in methods: 

89 suffix_token = f"_{suffix}" 

90 if operation_id.endswith(suffix_token): 

91 operation["operationId"] = operation_id[: -len(suffix_token)] + f"_{method}" 

92 break 

93 

94 

95class SnapshotFragment(TypedDict): 

96 paths: ReadOnly[Mapping[str, Mapping[str, object]]] 

97 components: ReadOnly[Mapping[str, Mapping[str, object]]] 

98 

99 

100@dataclass(frozen=True, slots=True) 

101class SnapshotResult: 

102 fragments: Mapping[str, SnapshotFragment] 

103 skipped: tuple[str, ...] 

104 

105 

106def _register_feature(app: "FastAPI", feat: "LazyFeature") -> str | None: 

107 import importlib 

108 

109 try: 

110 feat.register_fn(app, importlib.import_module(feat.module_path)) 

111 except Exception as exc: 

112 sys.stderr.write(f"warning: skip {feat.name}: {exc}\n") 

113 return feat.name 

114 return None 

115 

116 

117def _feature_fragment(app: "FastAPI", feat: "LazyFeature", used_operation_ids: set[str]) -> SnapshotFragment | None: 

118 from fastapi.openapi.utils import get_openapi 

119 

120 from litellm.proxy.proxy_server import ensure_unique_openapi_operation_ids 

121 

122 feat_routes: Final = [r for r in app.routes if feat.matches(getattr(r, "path", ""))] 

123 if not feat_routes: 

124 return None 

125 _stabilize_multi_method_route_ids(feat_routes) 

126 full: Final = get_openapi(title=app.title, version=app.version, routes=feat_routes) 

127 paths: Final = full.get("paths", {}) 

128 _normalize_operation_ids(paths) 

129 for path_ops in paths.values(): 

130 for method, op in path_ops.items(): 

131 if isinstance(op, dict): 

132 operation_id = op.get("operationId") 

133 if isinstance(operation_id, str): 

134 for suffix in HTTP_METHOD_SUFFIXES: 

135 if operation_id.endswith(f"_{suffix}"): 

136 op["operationId"] = operation_id[: -len(suffix)] + method 

137 break 

138 op["tags"] = [feat.name] 

139 unique: Final = ensure_unique_openapi_operation_ids(full, used_operation_ids) 

140 return { 

141 "paths": paths, 

142 "components": {"schemas": unique.get("components", {}).get("schemas", {})}, 

143 } 

144 

145 

146def generate_snapshot() -> SnapshotResult: 

147 from litellm.proxy._lazy_features import LAZY_FEATURES 

148 from litellm.proxy.proxy_server import app 

149 

150 skipped: Final = tuple(name for feat in LAZY_FEATURES if (name := _register_feature(app, feat)) is not None) 

151 used_operation_ids: Final[set[str]] = set() 

152 fragments: Final = { 

153 feat.name: fragment 

154 for feat in LAZY_FEATURES 

155 if (fragment := _feature_fragment(app, feat, used_operation_ids)) is not None 

156 } 

157 return SnapshotResult(fragments=fragments, skipped=skipped) 

158 

159 

160def main(snapshot_file: Path = SNAPSHOT_FILE, generate: Callable[[], SnapshotResult] = generate_snapshot) -> int: 

161 result: Final = generate() 

162 if result.skipped: 

163 sys.stderr.write( 

164 f"error: {len(result.skipped)} feature(s) failed to import, so their fragments would vanish from the " 

165 f"snapshot: {', '.join(result.skipped)}\n" 

166 ) 

167 return 1 

168 snapshot_file.write_text(json.dumps(result.fragments, indent=2, sort_keys=True) + "\n") 

169 sys.stdout.write(f"wrote {len(result.fragments)} feature fragments to {snapshot_file}\n") 

170 return 0 

171 

172 

173if __name__ == "__main__": 173 ↛ 174line 173 didn't jump to line 174 because the condition on line 173 was never true

174 sys.exit(main())