Coverage for scripts / generate_openapi.py: 100.00%

1#!/usr/bin/env python3 

2"""Generate the committed OpenAPI documents for every GCO HTTP service. 

3 

4Each FastAPI application already knows its own schema; this script asks each one 

5for ``app.openapi()`` and writes it to ``docs/openapi/<service>.json`` so the 

6API surface is reviewable in diffs and consumable by client generators without 

7running a cluster. 

8 

9Usage:: 

10 

11 python scripts/generate_openapi.py # write the documents 

12 python scripts/generate_openapi.py --check # fail if anything is stale 

13 

14``--check`` is what CI runs: it regenerates in memory and compares, so a route 

15added without regenerating is caught in review rather than shipping a schema 

16that disagrees with the code. 

17 

18The applications are imported, not deployed. ``GCO_DEV_MODE`` is set so the 

19authentication middleware's constructor does not log a configuration error for a 

20missing signing secret, and no AWS call is made during import. 

21""" 

22 

23from __future__ import annotations 

24 

25import argparse 

26import json 

27import os 

28import sys 

29from pathlib import Path 

30from typing import Any 

31 

32REPO_ROOT = Path(__file__).resolve().parent.parent 

33OUTPUT_DIR = REPO_ROOT / "docs" / "openapi" 

34 

35#: Service name -> module exposing a module-level ``app``. The service name is 

36#: the filename stem of the generated document and matches the Kubernetes 

37#: Service name the application is served under. 

38SERVICE_NAMES: tuple[str, ...] = ( 

39 "manifest-processor", 

40 "health-monitor", 

41 "inference-proxy", 

42 "cost-monitor", 

43) 

44 

45#: Routes FastAPI adds for its own interactive documentation. They are real 

46#: routes but they are not part of GCO's API contract, and no API Gateway 

47#: forwards them, so they are excluded to keep the documents about the service. 

48_DOC_ROUTE_PATHS = frozenset({"/docs", "/docs/oauth2-redirect", "/redoc", "/openapi.json"}) 

49 

50 

51def load_apps() -> dict[str, Any]: 

52 """Return ``{service name: FastAPI app}`` for every GCO HTTP service. 

53 

54 The imports are literal rather than resolved through 

55 ``importlib.import_module``: there is no reason for this mapping to be 

56 dynamic, and a static import set is both easier to follow and impossible to 

57 redirect at a module that was never intended to be loaded here. 

58 

59 Importing does not start a server or call AWS. ``GCO_DEV_MODE`` is set first 

60 so the authentication middleware's constructor does not log a configuration 

61 error about the signing secret it does not need for schema generation. 

62 """ 

63 os.environ.setdefault("GCO_DEV_MODE", "true") 

64 if str(REPO_ROOT) not in sys.path: 

65 sys.path.insert(0, str(REPO_ROOT)) 

66 

67 from gco.services import cost_api, health_api, inference_api, manifest_api 

68 

69 apps = { 

70 "manifest-processor": manifest_api.app, 

71 "health-monitor": health_api.app, 

72 "inference-proxy": inference_api.app, 

73 "cost-monitor": cost_api.app, 

74 } 

75 assert tuple(apps) == SERVICE_NAMES, "SERVICE_NAMES is out of step with load_apps()" 

76 return apps 

77 

78 

79def build_document(app: Any) -> dict[str, Any]: 

80 """Return one application's OpenAPI document, minus FastAPI's doc routes. 

81 

82 Round-tripped through JSON so the returned document contains only plain 

83 types, matching exactly what is written to disk. 

84 """ 

85 document: dict[str, Any] = json.loads(json.dumps(app.openapi())) 

86 for path in _DOC_ROUTE_PATHS: 

87 document.get("paths", {}).pop(path, None) 

88 return document 

89 

90 

91def render(document: dict[str, Any]) -> str: 

92 """Serialize deterministically so regeneration produces a stable diff.""" 

93 return json.dumps(document, indent=2, sort_keys=True) + "\n" 

94 

95 

96def main(argv: list[str] | None = None) -> int: 

97 parser = argparse.ArgumentParser(description=__doc__.splitlines()[0]) 

98 parser.add_argument( 

99 "--check", 

100 action="store_true", 

101 help="Do not write; exit non-zero if any committed document is stale.", 

102 ) 

103 args = parser.parse_args(argv) 

104 

105 OUTPUT_DIR.mkdir(parents=True, exist_ok=True) 

106 stale: list[str] = [] 

107 

108 for service, app in sorted(load_apps().items()): 

109 rendered = render(build_document(app)) 

110 target = OUTPUT_DIR / f"{service}.json" 

111 current = target.read_text(encoding="utf-8") if target.is_file() else None 

112 

113 if args.check: 

114 if current != rendered: 

115 stale.append(service) 

116 state = "missing" if current is None else "stale" 

117 print(f"{target.relative_to(REPO_ROOT)}: {state}") 

118 continue 

119 

120 if current == rendered: 

121 print(f"{target.relative_to(REPO_ROOT)}: unchanged") 

122 else: 

123 target.write_text(rendered, encoding="utf-8") 

124 print(f"{target.relative_to(REPO_ROOT)}: written") 

125 

126 if stale: 

127 print( 

128 "\nRegenerate with: python scripts/generate_openapi.py", 

129 file=sys.stderr, 

130 ) 

131 return 1 

132 return 0 

133 

134 

135if __name__ == "__main__": 

136 raise SystemExit(main())