[V5] openbb-cli: Modular OpenAPI Refs (#7589)

* follow up to #7585

* add another codegen test
This commit is contained in:
Danglewood
2026-07-14 12:49:37 -07:00
committed by GitHub
parent 92744a33cd
commit 5b20e24352
13 changed files with 750 additions and 46 deletions

View File

@@ -9,6 +9,7 @@ CredentialLocation = Literal["query", "header"]
_CREDENTIAL_NAMES: frozenset[str] = frozenset(
{
"api_key",
"api_token",
"apikey",
"x_api_key",
"x_apikey",

View File

@@ -297,13 +297,38 @@ def _build_root_router(
package_name: str,
root_namespace: str,
sub_routers: list[GeneratedRouter],
root_commands: list[str] | None = None,
root_post_imports: list[tuple[str, str]] | None = None,
root_has_stream: bool = False,
) -> GeneratedRouter:
"""Emit a top-level router that mounts every namespace under ``root_namespace``."""
"""Emit a top-level router that mounts every namespace under ``root_namespace``.
Top-level leaf commands (no namespace of their own) are rendered directly
on this router; unused imports are cleaned up by the ruff pass.
"""
commands = root_commands or []
post_imports = root_post_imports or []
parts: list[str] = [
f'"""Root router for {root_namespace} — generated from spec."""',
"",
"from openbb_core.app.router import Router",
]
if commands:
parts.append("from openbb_core.app.model.command_context import CommandContext")
parts.append("from openbb_core.app.model.obbject import OBBject")
if root_has_stream:
parts.append("from openbb_core.app.model.stream import OBBStream")
parts.append(
"from openbb_core.app.provider_interface import "
"ExtraParams, ProviderChoices, StandardParams"
)
parts.append("from openbb_core.app.query import Query")
parts.append("from openbb_core.app.router import Router")
if post_imports:
parts.append("")
for module_path, function_name in post_imports:
parts.append(
f"from {module_path} import {function_name} as _{function_name}"
)
if sub_routers:
parts.append("")
for r in sub_routers:
@@ -321,6 +346,9 @@ def _build_root_router(
f"router.include_router(_{r.module_name}_router, "
f'prefix="/{r.module_name}")'
)
if commands:
parts.append("")
parts.extend(commands)
return GeneratedRouter(
module_name=root_namespace,
entry_point_name=root_namespace,
@@ -466,6 +494,16 @@ def generate_packages(
f"{package_name}.providers..models.",
f"{package_name}.providers.tools.models.",
)
routers.root_post_imports = [
(
module_path.replace(
f"{package_name}.providers..models.",
f"{package_name}.providers.tools.models.",
),
function_name,
)
for module_path, function_name in routers.root_post_imports
]
root_namespace = _slugify(provider_name or project_slug)
sub_routers = [r for r in routers.routers if r.entry_point_name]
@@ -475,6 +513,9 @@ def generate_packages(
package_name=package_name,
root_namespace=root_namespace,
sub_routers=sub_routers,
root_commands=routers.root_commands,
root_post_imports=routers.root_post_imports,
root_has_stream=routers.root_has_stream,
)
routers.routers.append(root_router)

View File

@@ -40,9 +40,19 @@ class GeneratedRouters:
----------
routers : list of GeneratedRouter
All emitted router modules.
root_commands : list of str
Rendered command blocks for top-level leaf commands; they have no
namespace router of their own and mount directly on the root router.
root_post_imports : list of (str, str)
``(module_path, function_name)`` imports the root command blocks need.
root_has_stream : bool
Whether any root command streams (root router imports ``OBBStream``).
"""
routers: list[GeneratedRouter] = field(default_factory=list)
root_commands: list[str] = field(default_factory=list)
root_post_imports: list[tuple[str, str]] = field(default_factory=list)
root_has_stream: bool = False
def _safe_segment(name: str) -> str:
@@ -107,15 +117,42 @@ def generate_routers(
"""
out = GeneratedRouters()
for top_name, top_node in sorted(root.children.items()):
_emit_router(
top_node,
package_name=package_name,
provider_name=provider_name,
fetchers_by_command=fetchers_by_command,
post_commands_by_command=post_commands_by_command,
collected=out,
is_top_level=True,
)
if top_node.is_namespace:
_emit_router(
top_node,
package_name=package_name,
provider_name=provider_name,
fetchers_by_command=fetchers_by_command,
post_commands_by_command=post_commands_by_command,
collected=out,
is_top_level=True,
)
if top_node.cmd_spec is None:
continue
function_name = _safe_segment(top_name)
description = (top_node.cmd_spec.get("description") or "").strip()
fetcher = fetchers_by_command.get(top_node.full_path)
if fetcher is not None:
renderer = (
_render_stream_command if fetcher.is_streaming else _render_get_command
)
out.root_has_stream = out.root_has_stream or fetcher.is_streaming
out.root_commands.append(
renderer(fetcher, function_name=function_name, description=description)
)
continue
post = post_commands_by_command.get(top_node.full_path)
if post is not None:
out.root_post_imports.append(
(
f"{package_name}.providers.{provider_name}.models."
f"{post.module_name}",
post.function_name,
)
)
out.root_commands.append(
f'router.command(methods=["POST"])(_{post.function_name})\n'
)
return out