diff --git a/Makefile b/Makefile index 04e6e246c2..60ef245294 100644 --- a/Makefile +++ b/Makefile @@ -106,8 +106,7 @@ generate-cli-commands: ## Run generation of the CLI commands .PHONY: generate-cli-reference-docs generate-cli-reference-docs: ## Generate the CLI reference documentation - $(UV) run --frozen packages/nemo_platform_ext/scripts/docs_generator.py reference > docs/cli/reference.mdx - $(UV) run --frozen packages/nemo_platform_ext/scripts/docs_generator.py summary > docs/fern/snippets/_snippets/cli-summary.mdx + NMP_CONFIG_FILE_PATH="$(NMP_CONFIG_FILE_PATH)" $(UV) run --frozen packages/nemo_platform_ext/scripts/docs_generator.py all .PHONY: generate-config-reference-docs generate-config-reference-docs: ## Generate the platform config reference documentation diff --git a/packages/nemo_platform_ext/scripts/docs_generator.py b/packages/nemo_platform_ext/scripts/docs_generator.py index d88b7583e1..4d9071d691 100644 --- a/packages/nemo_platform_ext/scripts/docs_generator.py +++ b/packages/nemo_platform_ext/scripts/docs_generator.py @@ -16,6 +16,7 @@ from datetime import datetime from functools import cache from importlib import import_module +from pathlib import Path from types import ModuleType from typing import Any @@ -724,6 +725,11 @@ def _escape_mdx_line(line: str) -> str: } +_REPO_ROOT = Path(__file__).resolve().parents[3] +_REFERENCE_DOCS_PATH = _REPO_ROOT / "docs/cli/reference.mdx" +_SUMMARY_DOCS_PATH = _REPO_ROOT / "docs/fern/snippets/_snippets/cli-summary.mdx" + + def _enable_plugin_cli_docs() -> None: """Include supported plugin commands in generated CLI documentation.""" import os @@ -731,31 +737,46 @@ def _enable_plugin_cli_docs() -> None: os.environ.update(_PLUGIN_DOCS_DISCOVERY_ENV) +def _with_trailing_newline(output: str) -> str: + """Return output with exactly the trailing newline expected in generated files.""" + return output if output.endswith("\n") else output + "\n" + + +def write_docs_files(app: typer.Typer, reference_path: Path, summary_path: Path, name: str = "nemo") -> None: + """Write generated CLI reference and summary docs from one imported CLI app.""" + reference_path.parent.mkdir(parents=True, exist_ok=True) + summary_path.parent.mkdir(parents=True, exist_ok=True) + reference_path.write_text(_with_trailing_newline(generate_docs(app, name=name)), encoding="utf-8") + summary_path.write_text(_with_trailing_newline(generate_index_snippet(app, name=name)), encoding="utf-8") + + def main() -> None: """Generate CLI documentation and print to stdout. Usage: docs_generator.py reference # Full CLI reference docs_generator.py summary # Index page summary snippet + docs_generator.py all # Write both generated docs files """ import sys + if len(sys.argv) != 2 or sys.argv[1] not in ("reference", "summary", "all"): + print("Usage: docs_generator.py {reference|summary|all}", file=sys.stderr) + sys.exit(1) + _enable_plugin_cli_docs() from nemo_platform_ext.cli.app import app - if len(sys.argv) != 2 or sys.argv[1] not in ("reference", "summary"): - print("Usage: docs_generator.py {reference|summary}", file=sys.stderr) - sys.exit(1) - mode = sys.argv[1] + if mode == "all": + write_docs_files(app, _REFERENCE_DOCS_PATH, _SUMMARY_DOCS_PATH, name="nemo") + return if mode == "summary": output = generate_index_snippet(app, name="nemo") else: output = generate_docs(app, name="nemo") - sys.stdout.write(output) - if not output.endswith("\n"): - sys.stdout.write("\n") + sys.stdout.write(_with_trailing_newline(output)) if __name__ == "__main__": diff --git a/packages/nemo_platform_ext/tests/cli/test_docs_generator.py b/packages/nemo_platform_ext/tests/cli/test_docs_generator.py index 5021d11df2..0dba63aee9 100644 --- a/packages/nemo_platform_ext/tests/cli/test_docs_generator.py +++ b/packages/nemo_platform_ext/tests/cli/test_docs_generator.py @@ -30,6 +30,8 @@ def _load_docs_generator(): _docs_generator = _load_docs_generator() generate_docs = _docs_generator.generate_docs generate_index_snippet = _docs_generator.generate_index_snippet +write_docs_files = _docs_generator.write_docs_files +with_trailing_newline = _docs_generator._with_trailing_newline enable_plugin_cli_docs = _docs_generator._enable_plugin_cli_docs documented_plugin_clis = _docs_generator._DOCUMENTED_PLUGIN_CLIS plugin_docs_discovery_env = _docs_generator._PLUGIN_DOCS_DISCOVERY_ENV @@ -143,3 +145,25 @@ def visible() -> None: assert "hidden-command" not in snippet assert "Hidden command." not in snippet assert "* `--help, -h`: Show this message and exit." in reference + + +def test_write_docs_files_matches_individual_generators(tmp_path): + docs_app = typer.Typer() + + @docs_app.callback() + def main() -> None: + """Test CLI.""" + + @docs_app.command(rich_help_panel="Setup") + def visible() -> None: + """Visible command.""" + + reference_path = tmp_path / "docs/cli/reference.mdx" + summary_path = tmp_path / "docs/fern/snippets/_snippets/cli-summary.mdx" + + write_docs_files(docs_app, reference_path, summary_path, name="nemo") + + assert reference_path.read_text(encoding="utf-8") == generate_docs(docs_app, name="nemo") + assert summary_path.read_text(encoding="utf-8") == with_trailing_newline( + generate_index_snippet(docs_app, name="nemo") + ) diff --git a/script/generate-openapi-spec.sh b/script/generate-openapi-spec.sh index 7500b919f8..cc48cf40cc 100755 --- a/script/generate-openapi-spec.sh +++ b/script/generate-openapi-spec.sh @@ -2,4 +2,4 @@ # SPDX-FileCopyrightText: Copyright (c) 2025-2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. # SPDX-License-Identifier: Apache-2.0 -uv run --frozen python -m script.generate_openapi_spec -v "$@" +uv run --frozen python -m script.generate_openapi_spec "$@" diff --git a/script/generate_openapi_spec.py b/script/generate_openapi_spec.py index 99e3233389..7e8fc2b4c5 100644 --- a/script/generate_openapi_spec.py +++ b/script/generate_openapi_spec.py @@ -2,18 +2,25 @@ # SPDX-FileCopyrightText: Copyright (c) 2025-2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. # SPDX-License-Identifier: Apache-2.0 +from __future__ import annotations + import argparse import contextlib +import copy import importlib.metadata import inspect import json +import multiprocessing import os +import queue import shutil import sys +import time import traceback from concurrent.futures import ProcessPoolExecutor, as_completed from dataclasses import dataclass from enum import Enum +from io import StringIO from pathlib import Path from typing import Dict, List, Optional @@ -54,6 +61,7 @@ # Global verbose flag VERBOSE = False +PLUGIN_WORKER_TIMEOUT_SECONDS = 300 def print_green(message: str, verbose_only: bool = False): @@ -74,6 +82,60 @@ def print_verbose(message: str): print(message) +def _emit_worker_output(output: str, force: bool = False) -> None: + """Print captured subprocess output for verbose runs and failures.""" + if output and (VERBOSE or force): + print(output, end="" if output.endswith("\n") else "\n") + + +def _capture_generation_output(fn, item, verbose: bool) -> tuple[str, bool, str, str]: + """Run an OpenAPI extraction worker while capturing noisy import-time output.""" + global VERBOSE + VERBOSE = verbose + set_verbose(verbose) + + output = StringIO() + with contextlib.redirect_stdout(output), contextlib.redirect_stderr(output): + name, success, error = fn(item) + return name, success, error, output.getvalue() + + +def extract_openapi_spec_captured(service: ServiceConfig, verbose: bool) -> tuple[str, bool, str, str]: + return _capture_generation_output(extract_openapi_spec, service, verbose) + + +def extract_plugin_openapi_spec_captured(plugin: PluginConfig, verbose: bool) -> tuple[str, bool, str, str]: + return _capture_generation_output(extract_plugin_openapi_spec, plugin, verbose) + + +def bounded_worker_count( + item_count: int, + requested_workers: int | None = None, + default_limit: int = 4, + *, + cap_by_cpu: bool = True, +) -> int: + """Return a conservative process count for isolated OpenAPI worker pools.""" + if item_count <= 0: + return 0 + if requested_workers is not None: + if requested_workers < 1: + msg = "--plugin-workers must be at least 1" + raise ValueError(msg) + return min(item_count, requested_workers) + + cpu_count = os.cpu_count() or 1 + if cap_by_cpu: + return min(item_count, default_limit, cpu_count) + return min(item_count, default_limit) + + +def plugin_multiprocessing_context(): + """Use the cheapest process start method that preserves plugin isolation.""" + start_method = "fork" if "fork" in multiprocessing.get_all_start_methods() else "spawn" + return multiprocessing.get_context(start_method) + + class SpecType(Enum): GA = "ga" EA = "ea" @@ -238,9 +300,10 @@ def extract_openapi_specs_sequential(services: List[ServiceConfig]) -> None: for service in services: # Create a new executor for each service - this ensures a fresh process with ProcessPoolExecutor(max_workers=1) as executor: - future = executor.submit(extract_openapi_spec, service) + future = executor.submit(extract_openapi_spec_captured, service, VERBOSE) try: - name, success, error = future.result() + name, success, error, output = future.result() + _emit_worker_output(output, force=not success) if success: print_green(f"Completed: {name}") else: @@ -278,14 +341,17 @@ def extract_openapi_specs_with_process_pool(services: List[ServiceConfig]) -> No # ProcessPoolExecutor isolates imports in separate processes with ProcessPoolExecutor() as executor: # Submit all tasks - future_to_service = {executor.submit(extract_openapi_spec, service): service for service in services} + future_to_service = { + executor.submit(extract_openapi_spec_captured, service, VERBOSE): service for service in services + } # Collect results failed_services = [] for future in as_completed(future_to_service): service = future_to_service[future] try: - name, success, error = future.result() + name, success, error, output = future.result() + _emit_worker_output(output, force=not success) if success: print_green(f"Completed: {name}") else: @@ -428,35 +494,115 @@ def _extract_plugin_openapi_spec(plugin: PluginConfig) -> tuple[str, bool, str]: return plugin.dir, True, "" -def extract_plugin_specs_with_process_pool(plugins: List[PluginConfig]) -> None: +def extract_plugin_openapi_spec_to_queue(plugin: PluginConfig, verbose: bool, result_queue) -> None: + """Extract one plugin spec in a child process and return a captured result.""" + try: + result_queue.put(extract_plugin_openapi_spec_captured(plugin, verbose)) + except BaseException as exc: + error_msg = f"Plugin worker for {plugin.dir} crashed: {exc}\n{traceback.format_exc()}" + result_queue.put((plugin.dir, False, error_msg, "")) + + +def extract_plugin_spec_batch(plugin_batch: list[PluginConfig]) -> list[tuple[str, bool, str, str]]: + """Extract a batch with exactly one child process per plugin.""" + context = plugin_multiprocessing_context() + processes = [] + for plugin in plugin_batch: + result_queue = context.Queue(maxsize=1) + process = context.Process( + target=extract_plugin_openapi_spec_to_queue, + args=(plugin, VERBOSE, result_queue), + ) + process.start() + deadline = time.monotonic() + PLUGIN_WORKER_TIMEOUT_SECONDS + processes.append((plugin, process, result_queue, deadline)) + + results = [] + for plugin, process, result_queue, deadline in processes: + try: + while True: + remaining = deadline - time.monotonic() + if remaining <= 0: + if process.is_alive(): + process.kill() + process.join() + name = plugin.dir + success = False + output = "" + error = f"Plugin worker for {plugin.dir} timed out after {PLUGIN_WORKER_TIMEOUT_SECONDS}s" + else: + process.join() + try: + name, success, error, output = result_queue.get(timeout=1.0) + except queue.Empty: + name = plugin.dir + success = False + output = "" + error = ( + f"Plugin worker for {plugin.dir} exited with code {process.exitcode} " + "without returning a result" + ) + break + + try: + name, success, error, output = result_queue.get(timeout=min(0.1, remaining)) + break + except queue.Empty: + if not process.is_alive(): + process.join() + try: + name, success, error, output = result_queue.get(timeout=1.0) + except queue.Empty: + name = plugin.dir + success = False + output = "" + error = ( + f"Plugin worker for {plugin.dir} exited with code {process.exitcode} " + "without returning a result" + ) + break + process.join() + finally: + result_queue.close() + result_queue.join_thread() + + if process.exitcode not in (0, None) and success: + success = False + error = f"Plugin worker for {plugin.dir} exited with code {process.exitcode}" + results.append((name, success, error, output)) + + return results + + +def extract_plugin_specs_with_process_pool(plugins: List[PluginConfig], max_workers: int | None = None) -> None: """Extract OpenAPI specs for plugins via isolated subprocesses. - Each plugin gets a fresh ProcessPoolExecutor(max_workers=1) so a worker never - processes two plugins in one process. That matters on Linux (fork): discovering - services for plugin A imports route modules for plugin B, which registers - query-param filter schemas at import time; plugin B's extraction then calls - clear_query_param_schemas() and openapi() without re-importing those routes, - leaving dangling ``#/components/schemas/*Filter`` refs (e.g. MetricFilter). + Each plugin runs in a process that handles at most one plugin. That matters + on Linux: discovering services for plugin A can import route modules for + plugin B, which registers query-param filter schemas at import time; plugin + B's extraction then calls clear_query_param_schemas() and openapi() without + re-importing those routes, leaving dangling ``#/components/schemas/*Filter`` + refs (e.g. MetricFilter). ``max_tasks_per_child=1`` preserves that isolation + while still allowing multiple plugins to generate concurrently. """ - print_green(f"=== Generating OpenAPI specs for {len(plugins)} plugin(s) using process pool ===") - if not plugins: return + worker_count = bounded_worker_count(len(plugins), max_workers, default_limit=12, cap_by_cpu=False) + print_green( + f"=== Generating OpenAPI specs for {len(plugins)} plugin(s) using {worker_count} isolated worker(s) ===" + ) + failed_plugins = [] - for plugin in plugins: - with ProcessPoolExecutor(max_workers=1) as executor: - future = executor.submit(extract_plugin_openapi_spec, plugin) - try: - name, success, error = future.result() - if success: - print_green(f"Completed: {name}") - else: - failed_plugins.append((name, error)) - print_red(f"Failed: {name}") - except Exception as e: - failed_plugins.append((plugin.dir, str(e))) - print_red(f"Exception in {plugin.dir}: {str(e)}") + for start in range(0, len(plugins), worker_count): + plugin_batch = plugins[start : start + worker_count] + for name, success, error, output in extract_plugin_spec_batch(plugin_batch): + _emit_worker_output(output, force=not success) + if success: + print_green(f"Completed: {name}") + else: + failed_plugins.append((name, error)) + print_red(f"Failed: {name}") if failed_plugins: print_red(f"\n{len(failed_plugins)} plugins failed:") @@ -467,9 +613,26 @@ def extract_plugin_specs_with_process_pool(plugins: List[PluginConfig]) -> None: print_green(f"All {len(plugins)} plugin(s) completed successfully!") -def apply_schema_fixes(spec_files: List[str], apply_reorder: bool = True) -> None: - """Apply schema fixes to a list of OpenAPI spec files.""" - print_green("=== Applying fixes to OpenAPI schemas ===") +def apply_standard_schema_fixes(spec: dict, apply_reorder: bool = True) -> dict: + """Apply common schema normalization after path-specific edits.""" + spec = tweak_spec(spec) + spec = hoist_nested_defs(spec) + spec = remove_unused_schemas(spec) + spec = remove_invalid_components(spec) + spec = fix_recursive_schemas(spec) + spec = update_object_type(spec) + spec = mark_direct_span_json_value_for_stainless(spec) + spec["openapi"] = "3.1.0" + spec["info"]["version"] = platform_api_version + + if apply_reorder: + spec = reorder_spec(spec) + + return spec + + +def apply_schema_fixes_to_spec(spec: dict, spec_file: str, apply_reorder: bool = True) -> dict: + """Apply schema fixes to one OpenAPI spec.""" # Endpoints stripped from the public OpenAPI spec (not exposed in SDK). # Includes health/status and internal endpoints. health_endpoints = [ @@ -480,57 +643,51 @@ def apply_schema_fixes(spec_files: List[str], apply_reorder: bool = True) -> Non "/health/ready", ] + # Hoist nested `$defs` up front so every downstream pass (including + # remove_endpoint → remove_unused_schemas → build_schema_tree) can resolve + # refs like ``#/components/schemas/DatetimeFilter`` against top-level + # components instead of hunting through inline ``$defs``. + spec = hoist_nested_defs(spec) + for endpoint in health_endpoints: + remove_endpoint(spec, endpoint, prune_unused=False) + + # Special handling for deployment management + if "platform" in spec_file: + print_verbose(f"Applying deployment management specific fixes to {spec_file}") + # Rename schema + if "components" in spec and "schemas" in spec["components"]: + schemas = spec["components"]["schemas"] + if "PageResponse" in schemas: + schemas["DeploymentsPage"] = schemas.pop("PageResponse") + rename_schema_references(spec, "PageResponse", "DeploymentsPage") + + # Remove endpoints + remove_endpoint(spec, "/v1/deployments", prune_unused=False) + remove_endpoint(spec, "/v1/deployments/{deploymentId}", prune_unused=False) + + if "platform" in spec_file: + # Remove internal endpoints (not part of public API) + internal_endpoints = [p for p in spec.get("paths", {}).keys() if p.startswith("/internal/")] + for endpoint in internal_endpoints: + remove_endpoint(spec, endpoint, prune_unused=False) + + # Apply streaming fixes for specific files + if any(name in spec_file for name in ["platform"]): + print_verbose(f"Applying streaming fixes to {spec_file}") + spec = fix_openai_streaming_endpoints(spec) + + return apply_standard_schema_fixes(spec, apply_reorder=apply_reorder) + + +def apply_schema_fixes(spec_files: List[str], apply_reorder: bool = True) -> None: + """Apply schema fixes to a list of OpenAPI spec files.""" + print_green("=== Applying fixes to OpenAPI schemas ===") + for spec_file in spec_files: if os.path.exists(spec_file): print_verbose(f"Fixing schema for {spec_file}") spec = load_openapi_spec(spec_file) - # Hoist nested `$defs` up front so every downstream pass (including - # remove_endpoint → remove_unused_schemas → build_schema_tree) can - # resolve refs like ``#/components/schemas/DatetimeFilter`` against - # top-level components instead of hunting through inline ``$defs``. - spec = hoist_nested_defs(spec) - for endpoint in health_endpoints: - remove_endpoint(spec, endpoint) - - # Special handling for deployment management - if "platform" in spec_file: - print_verbose(f"Applying deployment management specific fixes to {spec_file}") - # Rename schema - if "components" in spec and "schemas" in spec["components"]: - schemas = spec["components"]["schemas"] - if "PageResponse" in schemas: - schemas["DeploymentsPage"] = schemas.pop("PageResponse") - rename_schema_references(spec, "PageResponse", "DeploymentsPage") - - # Remove endpoints - remove_endpoint(spec, "/v1/deployments") - remove_endpoint(spec, "/v1/deployments/{deploymentId}") - - if "platform" in spec_file: - # Remove internal endpoints (not part of public API) - internal_endpoints = [p for p in spec.get("paths", {}).keys() if p.startswith("/internal/")] - for endpoint in internal_endpoints: - remove_endpoint(spec, endpoint) - - # Apply streaming fixes for specific files - if any(name in spec_file for name in ["platform"]): - print_verbose(f"Applying streaming fixes to {spec_file}") - spec = fix_openai_streaming_endpoints(spec) - - # Apply the standard fix-schema logic - spec = tweak_spec(spec) - spec = hoist_nested_defs(spec) - spec = remove_unused_schemas(spec) - spec = remove_invalid_components(spec) - spec = fix_recursive_schemas(spec) - spec = update_object_type(spec) - spec = mark_direct_span_json_value_for_stainless(spec) - spec["openapi"] = "3.1.0" - spec["info"]["version"] = platform_api_version - - if apply_reorder: - spec = reorder_spec(spec) - + spec = apply_schema_fixes_to_spec(spec, spec_file, apply_reorder=apply_reorder) save_openapi_spec(spec, spec_file) @@ -569,28 +726,33 @@ def merge_and_process_specs() -> None: save_openapi_spec(merged_ea, "openapi/ea/openapi.yaml") +def apply_schema_removals_to_spec(spec: dict) -> dict: + """Apply schema removals to fix inconsistencies.""" + # Remove schemas and update references + schema_removals = [ + # ("DeploymentConfigOutput", "DeploymentConfig"), + # ("GuardrailConfigOutput", "GuardrailConfig"), + # ("EvaluationConfig", "EvaluationConfigOutput"), + # ("EvaluationTarget", "EvaluationTargetOutput"), + # ("CustomizationTarget", "CustomizationTargetOutput"), + ] + + for old_name, new_name in schema_removals: + if "components" in spec and "schemas" in spec["components"]: + schemas = spec["components"]["schemas"] + if old_name in schemas: + del schemas[old_name] + rename_schema_references(spec, old_name, new_name) + + return spec + + def apply_schema_removals() -> None: """Apply schema removals to fix inconsistencies.""" ga_spec_file = "openapi/ga/openapi.yaml" if os.path.exists(ga_spec_file): spec = load_openapi_spec(ga_spec_file) - - # Remove schemas and update references - schema_removals = [ - # ("DeploymentConfigOutput", "DeploymentConfig"), - # ("GuardrailConfigOutput", "GuardrailConfig"), - # ("EvaluationConfig", "EvaluationConfigOutput"), - # ("EvaluationTarget", "EvaluationTargetOutput"), - # ("CustomizationTarget", "CustomizationTargetOutput"), - ] - - for old_name, new_name in schema_removals: - if "components" in spec and "schemas" in spec["components"]: - schemas = spec["components"]["schemas"] - if old_name in schemas: - del schemas[old_name] - rename_schema_references(spec, old_name, new_name) - + spec = apply_schema_removals_to_spec(spec) save_openapi_spec(spec, ga_spec_file) @@ -615,7 +777,7 @@ def apply_final_fixes() -> None: apply_schema_fixes(FINAL_SPEC_FILES) -def remove_guardrail_endpoints() -> None: +def remove_guardrail_endpoints_from_spec(spec: dict) -> dict: """Remove guardrail models endpoints from all final specs.""" guardrail_endpoints = [ @@ -623,26 +785,21 @@ def remove_guardrail_endpoints() -> None: ("/v2/guardrail/models/{model_id}", None), ] + # Remove guardrail endpoints + for endpoint, method in guardrail_endpoints: + remove_endpoint(spec, endpoint, method, prune_unused=False) + + # Apply schema fixes after removals, but do not reorder to preserve tag ordering. + return apply_standard_schema_fixes(spec, apply_reorder=False) + + +def remove_guardrail_endpoints() -> None: + """Remove guardrail models endpoints from all final specs.""" + for spec_file in FINAL_SPEC_FILES: if os.path.exists(spec_file): spec = load_openapi_spec(spec_file) - - # Remove guardrail endpoints - for endpoint, method in guardrail_endpoints: - remove_endpoint(spec, endpoint, method) - - # Apply schema fixes after removals (but don't reorder to preserve tag ordering) - spec = tweak_spec(spec) - spec = hoist_nested_defs(spec) - spec = remove_unused_schemas(spec) - spec = remove_invalid_components(spec) - spec = fix_recursive_schemas(spec) - spec = update_object_type(spec) - spec = mark_direct_span_json_value_for_stainless(spec) - spec["openapi"] = "3.1.0" - spec["info"]["version"] = platform_api_version - # Note: Don't call reorder_spec() here to preserve tag-based ordering - + spec = remove_guardrail_endpoints_from_spec(spec) save_openapi_spec(spec, spec_file) @@ -732,7 +889,66 @@ def validate_final_specs(spec_files: List[str]) -> None: raise RuntimeError(f"{sum(len(d) for _, d in all_dangling)} dangling $refs detected") -def process_plugin_specs() -> None: +def can_process_single_platform_spec_in_memory(services: list[ServiceConfig]) -> bool: + """Return true when platform outputs are known to be identical.""" + if len(services) != 1: + return False + + service = services[0] + if not service.is_ga() or service.final_output_path() is None or service.copy_from: + return False + + # Tags/examples are final-spec-only transformations in the generic path. + # Keep that path if those inputs exist so individual and aggregate outputs + # retain their existing semantics. + return ( + not Path("openapi/ea/openapi.yaml").exists() + and not Path("openapi/nmp-common.openapi.yaml").exists() + and not any(Path("openapi/api-examples").glob("*.json")) + ) + + +def process_single_platform_spec_in_memory(services: list[ServiceConfig]) -> bool: + """Process the current single-platform OpenAPI layout without repeated file passes.""" + if not can_process_single_platform_spec_in_memory(services): + return False + + service = services[0] + temp_path = service.temp_output_path() + final_path = service.final_output_path() + if final_path is None or not os.path.exists(temp_path): + return False + + print_green("=== Processing single platform OpenAPI spec in memory ===") + spec = load_openapi_spec(temp_path) + spec = apply_schema_fixes_to_spec(spec, temp_path) + individual_spec = fix_ref_with_additional_props(copy.deepcopy(spec)) + + final_spec = apply_schema_removals_to_spec(spec) + final_spec = apply_schema_fixes_to_spec(final_spec, "openapi/openapi.yaml") + final_spec = remove_guardrail_endpoints_from_spec(final_spec) + final_spec = fix_ref_with_additional_props(final_spec) + + dangling_specs = [ + ("platform individual OpenAPI spec", validate_refs(individual_spec)), + ("platform OpenAPI spec", validate_refs(final_spec)), + ] + for spec_name, dangling in dangling_specs: + if dangling: + print_red(f"Found dangling $refs in the {spec_name}:") + for ref in dangling: + print_red(f" - {ref}") + raise RuntimeError(f"{len(dangling)} dangling $refs detected") + + for output_path in ["openapi/openapi.yaml", "openapi/ga/openapi.yaml"]: + save_openapi_spec(final_spec, output_path) + save_openapi_spec(individual_spec, final_path) + + os.remove(temp_path) + return True + + +def process_plugin_specs(plugin_workers: int | None = None) -> None: """Generate, fix, and validate OpenAPI specs for all opted-in plugins. Plugin specs land in ``plugins//openapi/`` and are never merged into @@ -743,7 +959,7 @@ def process_plugin_specs() -> None: return print_green(f"=== STEP 5: Generating OpenAPI specs for {len(plugins)} plugin(s) ===") - extract_plugin_specs_with_process_pool(plugins) + extract_plugin_specs_with_process_pool(plugins, max_workers=plugin_workers) plugin_spec_files = [p.output_path() for p in plugins] @@ -779,6 +995,12 @@ def main(): parser.add_argument( "--only-gen-schema", action="store_true", help="Only generate the schema, then exist, for debugging purposes" ) + parser.add_argument( + "--plugin-workers", + type=int, + default=None, + help="Maximum isolated worker processes for plugin OpenAPI generation (default: min(12, plugins))", + ) args = parser.parse_args() # Set global verbose flag @@ -823,32 +1045,33 @@ def main(): if args.only_gen_schema: return - apply_schema_fixes(all_spec_files) + if not process_single_platform_spec_in_memory(services_to_generate): + apply_schema_fixes(all_spec_files) - # Merge and process specs - use same logic for both modes - merge_and_process_specs() + # Merge and process specs - use same logic for both modes + merge_and_process_specs() - apply_schema_removals() + apply_schema_removals() - merge_final_specs() + merge_final_specs() - # Apply final fixes - use same logic for consistency - apply_final_fixes() + # Apply final fixes - use same logic for consistency + apply_final_fixes() - add_examples_and_finalize() + add_examples_and_finalize() - # Remove health endpoints and apply final processing (after tag ordering) - remove_guardrail_endpoints() + # Remove health endpoints and apply final processing (after tag ordering) + remove_guardrail_endpoints() - move_individual_specs() + move_individual_specs() - platform_spec_files = [ - path for path in (service.final_output_path() for service in SERVICES) if path is not None - ] + FINAL_SPEC_FILES - fix_ref_not_allowed_errors(platform_spec_files) - validate_final_specs(platform_spec_files) + platform_spec_files = [ + path for path in (service.final_output_path() for service in SERVICES) if path is not None + ] + FINAL_SPEC_FILES + fix_ref_not_allowed_errors(platform_spec_files) + validate_final_specs(platform_spec_files) - process_plugin_specs() + process_plugin_specs(plugin_workers=args.plugin_workers) print_green("=== OpenAPI spec generation completed successfully! ===") diff --git a/script/openapi_helper/openapi_tools.py b/script/openapi_helper/openapi_tools.py index afdee3cbda..4cac734c8c 100644 --- a/script/openapi_helper/openapi_tools.py +++ b/script/openapi_helper/openapi_tools.py @@ -6,6 +6,8 @@ """ import json +import os +import tempfile from copy import deepcopy from pathlib import Path from typing import Optional, Tuple @@ -53,12 +55,25 @@ def load_openapi_spec(file_path: str) -> dict: def save_openapi_spec(spec: dict, output_path: str) -> None: """Save OpenAPI specification to a YAML or JSON file.""" - if output_path.endswith(".json"): - with open(output_path, "w") as f: - f.write(json.dumps(spec, indent=2)) - else: - with open(output_path, "w") as f: - yaml.dump(spec, f, default_flow_style=False, sort_keys=False) + output = Path(output_path) + output.parent.mkdir(parents=True, exist_ok=True) + suffix = output.suffix or ".tmp" + existing_mode = output.stat().st_mode & 0o7777 if output.exists() else None + temp_path: Path | None = None + try: + with tempfile.NamedTemporaryFile("w", delete=False, dir=output.parent, suffix=suffix, encoding="utf-8") as f: + temp_path = Path(f.name) + if output_path.endswith(".json"): + f.write(json.dumps(spec, indent=2)) + else: + yaml.dump(spec, f, default_flow_style=False, sort_keys=False) + if existing_mode is not None: + os.chmod(temp_path, existing_mode) + os.replace(temp_path, output) + except Exception: + if temp_path is not None: + temp_path.unlink(missing_ok=True) + raise print_verbose(f"Saved specification to {output_path}", style="bold green") @@ -1034,7 +1049,7 @@ def remove_schema_command( raise typer.Exit(1) -def remove_endpoint(spec: dict, path: str, method: Optional[str] = None) -> None: +def remove_endpoint(spec: dict, path: str, method: Optional[str] = None, prune_unused: bool = True) -> None: """Remove an endpoint and its associated schemas from the specification. If method is None, removes all methods for the given path.""" if "paths" not in spec: @@ -1061,8 +1076,8 @@ def remove_endpoint(spec: dict, path: str, method: Optional[str] = None) -> None # Remove all methods for the path del spec["paths"][path] - # Remove any unused schemas - remove_unused_schemas(spec) + if prune_unused: + remove_unused_schemas(spec) @app.command(name="remove-endpoint") diff --git a/tools/lint/lint-fix-web-sdk.sh b/tools/lint/lint-fix-web-sdk.sh index 13981f1fc3..af67bcc17e 100755 --- a/tools/lint/lint-fix-web-sdk.sh +++ b/tools/lint/lint-fix-web-sdk.sh @@ -19,4 +19,4 @@ if ! command -v pnpm >/dev/null 2>&1; then fi pnpm_path="$(command -v pnpm)" -exec node "${pnpm_path}" gen +exec "${pnpm_path}" gen diff --git a/tools/lint/lint-web-sdk.sh b/tools/lint/lint-web-sdk.sh index 7554cf708f..9478c71459 100755 --- a/tools/lint/lint-web-sdk.sh +++ b/tools/lint/lint-web-sdk.sh @@ -20,4 +20,4 @@ if ! command -v pnpm >/dev/null 2>&1; then fi pnpm_path="$(command -v pnpm)" -exec node "${pnpm_path}" gen:check +exec "${pnpm_path}" gen:check diff --git a/web/packages/sdk/generateAll.ts b/web/packages/sdk/generateAll.ts index 56d0569763..3a3576e1ed 100644 --- a/web/packages/sdk/generateAll.ts +++ b/web/packages/sdk/generateAll.ts @@ -13,6 +13,7 @@ import crypto from 'crypto'; import { execSync } from 'child_process'; import fs from 'fs'; +import os from 'os'; import path from 'path'; import { serviceConfigs } from './orval/constants'; @@ -20,9 +21,38 @@ const FORCE = process.argv.includes('--force'); const GENERATED_DIR = path.join(__dirname, 'generated'); const HASH_FILE = path.join(GENERATED_DIR, '.input-hash'); const ORVAL_DIR = path.join(__dirname, 'orval'); +const MIN_SERVICE_PIPELINES = 4; +const MAX_PROCESSES_ENV = 'WEB_SDK_GEN_MAX_PROCESSES'; +const MAX_PROCESSES_PATTERN = /^(\d+)(%)?$/; const services = Object.keys(serviceConfigs) as Array; +const availableParallelism = (): number => + typeof os.availableParallelism === 'function' ? os.availableParallelism() : os.cpus().length; + +const defaultMaxProcesses = (): number => { + return Math.min(services.length, Math.max(MIN_SERVICE_PIPELINES, availableParallelism())); +}; + +const maxProcesses = (): string => { + const configured = process.env[MAX_PROCESSES_ENV]?.trim(); + if (configured) { + const match = configured.match(MAX_PROCESSES_PATTERN); + const value = match ? Number(match[1]) : Number.NaN; + if (!match || value < 1) { + throw new Error( + `${MAX_PROCESSES_ENV} must be a positive count or percent, got "${configured}"` + ); + } + + const configuredProcesses = match[2] + ? Math.round((availableParallelism() * value) / 100) + : value; + return String(Math.min(services.length, Math.max(1, configuredProcesses))); + } + return String(defaultMaxProcesses()); +}; + interface GenerationConfig { service: string; zod?: boolean; @@ -131,9 +161,12 @@ const main = async () => { const commands = generationConfigs.map(generateCommands); const serviceNames = generationConfigs.map((config) => config.service); + const maxServicePipelines = maxProcesses(); const colors = ['red', 'blue', 'green', 'yellow', 'magenta', 'cyan', 'purple', 'white', 'gray']; const concurrentlyArgs = [ + '--max-processes', + maxServicePipelines, '--names', serviceNames.join(','), '-c', @@ -143,7 +176,7 @@ const main = async () => { const concurrentlyCommand = `pnpm exec concurrently ${concurrentlyArgs.join(' ')}`; - console.log('Executing commands in parallel:'); + console.log(`Executing commands in parallel (max ${maxServicePipelines} service pipelines):`); commands.forEach((cmd, index) => { console.log(` ${index + 1}. ${cmd}`); });