"""Main CLI Module."""
import argparse
import contextlib
import difflib
import os
import re
import sys
import webbrowser
from datetime import datetime
from functools import partial, update_wrapper
from pathlib import Path
from types import MethodType
from typing import Any
import pandas as pd
from prompt_toolkit.formatted_text import HTML
from prompt_toolkit.styles import Style
from openbb_cli.backend import Backend, LocalBackend
from openbb_cli.config import constants
from openbb_cli.config.constants import (
ASSETS_DIRECTORY,
ENV_FILE_SETTINGS,
HOME_DIRECTORY,
REPOSITORY_DIRECTORY,
)
from openbb_cli.config.menu_text import MenuText
from openbb_cli.controllers.base_controller import BaseController
from openbb_cli.controllers.platform_controller_factory import (
PlatformControllerFactory,
)
from openbb_cli.controllers.script_parser import is_reset, parse_openbb_script
from openbb_cli.controllers.utils import (
bootup,
first_time_user,
get_flair_and_username,
parse_and_split_input,
print_goodbye,
print_rich_table,
reset,
suppress_stdout,
welcome_message,
)
from openbb_cli.session import Session
NON_DATA_ROUTERS = ["coverage", "reference", "system", "user"]
DATA_PROCESSING_ROUTERS = ["technical", "quantitative", "econometrics"]
env_file = str(ENV_FILE_SETTINGS)
session = Session()
def _result_to_obbject(result: Any, router: str, other_args: list[str]) -> Any:
"""Lift a spec-mode dispatcher result to an ``OBBject`` for registry insert."""
try:
from openbb_core.app.model.obbject import OBBject
except ImportError:
return None
if isinstance(result, OBBject):
obbject = result
elif isinstance(result, (list, dict)):
try:
obbject = OBBject(results=result)
obbject._route = "/" + router.replace(".", "/").strip("/")
except Exception: # noqa: BLE001
return None
else:
return None
obbject.extra["command"] = f"/{router} {' '.join(other_args)}".strip()
return obbject
def _params_to_completions(parameters: list[dict[str, Any]]) -> dict[str, Any]:
"""Translate normalized spec parameters into a NestedCompleter dict."""
out: dict[str, Any] = {}
for p in parameters or []:
name = p.get("name")
if not name:
continue
flag = f"--{name}"
choices_list = p.get("choices") or []
if choices_list:
out[flag] = {str(c): None for c in choices_list}
else:
out[flag] = None
out["--help"] = None
out["-h"] = "--help"
return out
class CLIController(BaseController):
"""CLI Controller class."""
CHOICES_COMMANDS_BUILTIN = ["record", "stop", "exe", "results"]
CHOICES_MENUS_BUILTIN = ["settings", "user", "feature"]
PATH = "/"
CHOICES_GENERATION = False
def __init__(
self,
jobs_cmds: list[str] | None = None,
*,
backend: Backend | None = None,
):
"""Construct CLI controller."""
self.ROUTINE_FILES: dict[str, Path] = dict()
self.ROUTINE_CHOICES: dict[str, Any] = dict()
self._backend: Backend = backend if backend is not None else LocalBackend()
platform_routers = self._backend.routers
menus = list(self.CHOICES_MENUS_BUILTIN)
commands = list(self.CHOICES_COMMANDS_BUILTIN)
for router, kind in platform_routers.items():
if router == "user":
continue
(menus if kind == "menu" else commands).append(router)
self.CHOICES_MENUS = menus
self.CHOICES_COMMANDS = commands
super().__init__(jobs_cmds)
self.queue: list[str] = list()
if jobs_cmds:
self.queue = parse_and_split_input(
an_input=" ".join(jobs_cmds), custom_filters=[]
)
self.update_success = False
self._generate_platform_commands()
self.update_runtime_choices()
def _generate_platform_commands(self):
"""Generate Platform based commands/menus from the configured backend."""
backend = self._backend
def method_call_class(self, _, controller, name, parent_path, target):
self.queue = self.load_class(
controller, name, parent_path, target, self.queue
)
def method_call_command_legacy(self, _, router: str):
"""Display the static command-target metadata for in-process ``obb``."""
mdl = backend.get_command_target(router)
df = pd.DataFrame.from_dict(mdl.model_dump(), orient="index")
if isinstance(df.columns, pd.RangeIndex):
df.columns = [str(i) for i in df.columns]
return print_rich_table(df, show_index=True)
def method_call_command_spec(self, other_args: list[str], router: str):
"""Dispatch a top-level command-typed router via SpecBackend."""
from openbb_cli.backend import SpecTranslator
spec_doc: dict[str, Any] = getattr(backend, "_spec", {})
cmd_spec = spec_doc.get("commands", {}).get(router)
if cmd_spec is None:
session.console.print(f"[red]Command not found in spec: {router}[/red]")
return
translator = SpecTranslator(
router, cmd_spec, getattr(backend, "_dispatcher", None)
)
ns_parser = self.parse_known_args_and_warn(
parser=translator.parser,
other_args=other_args,
export_allowed="raw_data_only",
)
if not ns_parser:
return
try:
result = translator.execute_func(ns_parser)
except Exception as exc: # noqa: BLE001
session.console.print(f"[red]error: {exc}[/red]")
return
obbject = _result_to_obbject(result, router, other_args)
if obbject is not None and obbject.results is not None:
if session.max_obbjects_exceeded():
session.obbject_registry.remove()
session.console.print(
"[yellow]Maximum number of OBBjects reached. "
"The oldest entry was removed.[/yellow]"
)
if session.obbject_registry.register(obbject) and (
session.settings.SHOW_MSG_OBBJECT_REGISTRY
):
session.console.print("Added `OBBject` to cached results.")
payload: Any
if obbject is not None:
payload = obbject.results
elif isinstance(result, dict):
payload = result.get("results", result)
else:
payload = result
if isinstance(payload, dict):
df = pd.DataFrame([payload])
elif isinstance(payload, list):
df = pd.DataFrame(payload)
else:
df = pd.DataFrame({"value": [payload]})
session.output_adapter.display(df, title=f"/{router}")
is_spec_backend = hasattr(backend, "_spec") and hasattr(backend, "_dispatcher")
method_call_command: Any = (
method_call_command_spec if is_spec_backend else method_call_command_legacy
)
spec_commands: dict[str, Any] = (
getattr(backend, "_spec", {}).get("commands", {}) if is_spec_backend else {}
)
def method_call_menu_or_leaf(
self,
other_args: list[str],
controller,
name: str,
parent_path: list[str],
target,
router: str,
):
"""Hybrid for routers that have both a leaf path and nested commands."""
if other_args:
method_call_command_spec(self, other_args, router)
return
self.queue = self.load_class(
controller, name, parent_path, target, self.queue
)
for router, value in backend.routers.items():
if router == "user":
continue
has_leaf = router in spec_commands
if value == "menu":
pcf = PlatformControllerFactory(backend=backend, router_name=router)
DynamicController = pcf.create()
if has_leaf:
bound_method = MethodType(method_call_menu_or_leaf, self)
bound_method = update_wrapper(
partial(
bound_method,
controller=DynamicController,
name=router,
target=None,
parent_path=self.path,
router=router,
),
method_call_menu_or_leaf,
)
else:
bound_method = MethodType(method_call_class, self)
bound_method = update_wrapper(
partial(
bound_method,
controller=DynamicController,
name=router,
target=None,
parent_path=self.path,
),
method_call_class,
)
else:
bound_method = MethodType(method_call_command, self)
bound_method = update_wrapper(
partial(bound_method, router=router),
method_call_command,
)
setattr(self, f"call_{router}", bound_method)
def _spec_command_completions(self) -> dict[str, dict[str, Any]]:
"""Build per-command completion choices from the SpecBackend's metadata."""
out: dict[str, dict[str, Any]] = {}
backend = getattr(self, "_backend", None)
if backend is None:
return out
spec_doc = getattr(backend, "_spec", None)
if not isinstance(spec_doc, dict):
return out
commands = spec_doc.get("commands", {})
for router in backend.routers:
if router not in self.controller_choices:
continue
cmd_spec = commands.get(router)
if not cmd_spec:
continue
out[router] = _params_to_completions(cmd_spec.get("parameters", []))
return out
def update_runtime_choices(self):
"""Update runtime choices."""
routines_directory = Path(session.user.preferences.export_directory, "routines")
if session.prompt_session and session.settings.USE_PROMPT_TOOLKIT:
choices: dict = {c: {} for c in self.controller_choices}
choices.update(self._spec_command_completions())
self.ROUTINE_FILES = {
filepath.name: filepath
for filepath in routines_directory.rglob("*.openbb")
}
choices["exe"] = {
"--file": {
filename: {} for filename in list(self.ROUTINE_FILES.keys())
},
"-f": "--file",
"--example": None,
"-e": "--example",
"--input": None,
"-i": "--input",
"--help": None,
"-h": "--help",
}
choices["record"] = {
"--name": None,
"-n": "--name",
"--description": None,
"-d": "--description",
"--tag1": {c: None for c in constants.SCRIPT_TAGS},
"--tag2": {c: None for c in constants.SCRIPT_TAGS},
"--tag3": {c: None for c in constants.SCRIPT_TAGS},
"--help": None,
"-h": "--help",
}
choices["stop"] = {"--help": None, "-h": "--help"}
registry_all = session.obbject_registry.all
index_completions = {str(idx): None for idx in registry_all}
key_completions = {
(data.get("extra") or {}).get("register_key"): None
for data in registry_all.values()
if (data.get("extra") or {}).get("register_key")
}
choices["results"] = {
"--index": index_completions if index_completions else None,
"-i": "--index",
"--key": key_completions if key_completions else None,
"-k": "--key",
"--chart": None,
"-c": "--chart",
"--export": {
c: None
for c in [
"csv",
"json",
"xlsx",
"png",
"jpg",
"db",
"sqlite",
"sqlite3",
]
},
"-e": "--export",
"--sheet-name": None,
"--help": None,
"-h": "--help",
}
choices["load"] = {
"--file": None,
"-f": "--file",
"--sheet-name": None,
"--register_key": None,
"--help": None,
"-h": "--help",
}
self.update_completer(choices)
def print_help(self):
"""Print help."""
mt = MenuText("")
mt.add_info("\nConfigure CLI")
mt.add_menu(
"settings",
description="enable and disable feature flags, preferences and settings",
)
mt.add_menu(
"user",
description="view and set platform user preferences for the session",
)
mt.add_raw("\n")
mt.add_info("Record and execute your own .openbb routine scripts")
mt.add_cmd("record", description="start recording current session")
mt.add_cmd(
"stop", description="stop session recording and convert to .openbb routine"
)
mt.add_cmd(
"exe",
description="execute .openbb routine scripts (use exe --example for an example)",
)
mt.add_raw("\n")
mt.add_info("Retrieve data from different asset classes and providers")
platform_routers = self._backend.routers
reference_routers = self._backend.reference_routers
reference_paths = self._backend.reference_paths
def _menu_description(router: str) -> str:
"""Return the menu's reference description (tolerant of trailing slash)."""
for key in (f"{self.PATH}{router}/", f"{self.PATH}{router}"):
desc = reference_routers.get(key, {}).get("description") or ""
if desc:
return desc.split(".")[0].lower()
return ""
def _command_description(router: str) -> str:
"""Return the operation description for top-level command-typed routers."""
for key in (f"{self.PATH}{router}", f"{self.PATH}{router}/"):
desc = reference_paths.get(key, {}).get("description") or ""
if desc:
return desc.split(".")[0].lower()
return ""
for router, value in platform_routers.items():
if router in NON_DATA_ROUTERS or router in DATA_PROCESSING_ROUTERS:
continue
if value == "menu":
mt.add_menu(name=router, description=_menu_description(router))
else:
mt.add_cmd(name=router, description=_command_description(router))
if any(router in platform_routers for router in DATA_PROCESSING_ROUTERS):
mt.add_info("\nAnalyze and process previously obtained data")
for router, value in platform_routers.items():
if router not in DATA_PROCESSING_ROUTERS:
continue
if value == "menu":
mt.add_menu(name=router, description=_menu_description(router))
else:
mt.add_cmd(name=router, description=_command_description(router))
mt.add_raw("\n")
mt.add_info("Data manipulation and feature engineering")
mt.add_menu("feature", description="feature engineering and table operations")
mt.add_cmd("load", description="load data from file in OpenBBUserData")
mt.add_cmd("results")
if session.obbject_registry.obbjects:
mt.add_info("\nCached Results")
for key, value in list(session.obbject_registry.all.items())[
: session.settings.N_TO_DISPLAY_OBBJECT_REGISTRY
]:
command = (value.get("extra") or {}).get("command", "")
mt.add_raw(
f"[yellow]OBB{key}[/yellow]: {command}",
left_spacing=True,
)
session.console.print(text=mt.menu_text, menu="Home")
self.update_runtime_choices()
def call_settings(self, _):
"""Process settings command."""
from openbb_cli.controllers.settings_controller import (
SettingsController,
)
self.queue = self.load_class(SettingsController, self.queue)
def call_user(self, _):
"""Process user command."""
from openbb_cli.controllers.user_controller import UserController
self.queue = self.load_class(UserController, self.queue)
def call_feature(self, _):
"""Process feature engineering command."""
from openbb_cli.controllers.feature_controller import (
FeatureController,
)
self.queue = self.load_class(FeatureController, self.queue)
def call_exe(self, other_args: list[str]):
"""Process exe command."""
other_args += self.queue
if not other_args:
session.console.print(
"[info]Provide a path to the routine you wish to execute. For an example, please use "
"`exe --example`.\n[/info]"
)
return
parser = argparse.ArgumentParser(
add_help=False,
formatter_class=argparse.ArgumentDefaultsHelpFormatter,
prog="exe",
description="Execute automated routine script. For an example, please use `exe --example`.",
)
parser.add_argument(
"--file",
"-f",
help="The path or .openbb file to run.",
dest="file",
required="-h" not in other_args
and "--help" not in other_args
and "-e" not in other_args
and "--example" not in other_args,
type=str,
nargs="+",
)
parser.add_argument(
"-i",
"--input",
help="Select multiple inputs to be replaced in the routine and separated by commas. E.g. GME,AMC,BTC-USD",
dest="routine_args",
type=str,
)
parser.add_argument(
"-e",
"--example",
help="Run an example script to understand how routines can be used.",
dest="example",
action="store_true",
default=False,
)
if other_args and "-" not in other_args[0][0]:
other_args.insert(0, "--file")
ns_parser = self.parse_known_args_and_warn(parser, other_args)
if ns_parser:
if ns_parser.example:
routine_path = ASSETS_DIRECTORY / "routines" / "routine_example.openbb"
session.console.print(
"[info]Executing an example, please visit our docs to learn how to create your own script.[/info]\n"
)
elif ns_parser.file:
file_path = " ".join(ns_parser.file)
routine_path = Path(self.ROUTINE_FILES.get(file_path, file_path))
else:
return
try:
with open(routine_path) as fp:
raw_lines = list(fp)
script_inputs = []
if routine_args := ns_parser.routine_args:
pattern = r"\[(.*?)\]"
matches = re.findall(pattern, routine_args)
for match in matches:
routine_args = routine_args.replace(f"[{match}]", "")
script_inputs.append(match)
script_inputs.extend(
[val for val in routine_args.split(",") if val]
)
err, parsed_script = parse_openbb_script(
raw_lines=raw_lines, script_inputs=script_inputs
)
if err:
session.console.print(err)
return
self.queue = [
val
for val in parse_and_split_input(
an_input=parsed_script, custom_filters=[]
)
if val
]
if "export" in self.queue[0]:
export_path = self.queue[0].split(" ")[1]
if export_path[0] == "~":
export_path = export_path.replace(
"~", HOME_DIRECTORY.as_posix()
)
elif export_path[0] != "/":
export_path = os.path.join(
os.path.dirname(os.path.abspath(__file__)), export_path
)
if os.path.isdir(export_path):
session.console.print(
f"Export data to be saved in the selected folder: '{export_path}'"
)
else:
os.makedirs(export_path)
session.console.print(
f"[green]Folder '{export_path}' successfully created.[/green]"
)
self.queue = self.queue[1:]
except FileNotFoundError:
session.console.print(
f"[red]File '{routine_path}' doesn't exist.[/red]"
)
return
def handle_job_cmds(jobs_cmds: list[str] | None) -> list[str] | None:
"""Handle job commands."""
export_path = ""
if jobs_cmds and "export" in jobs_cmds[0]:
commands = jobs_cmds[0].split("/")
first_split = commands[0].split(" ")
if len(first_split) > 1:
export_path = first_split[1]
jobs_cmds = ["/".join(commands[1:])]
if not export_path:
return jobs_cmds
if export_path[0] == "~":
export_path = export_path.replace("~", HOME_DIRECTORY.as_posix())
elif export_path[0] != "/":
export_path = os.path.join(
os.path.dirname(os.path.abspath(__file__)), export_path
)
if os.path.isdir(export_path):
session.console.print(
f"Export data to be saved in the selected folder: '{export_path}'"
)
else:
os.makedirs(export_path)
session.console.print(
f"[green]Folder '{export_path}' successfully created.[/green]"
)
return jobs_cmds
def run_cli(
jobs_cmds: list[str] | None = None,
test_mode=False,
*,
backend: Backend | None = None,
):
"""Run the CLI menu, optionally backed by a non-default ``Backend``."""
ret_code = 1
t_controller = CLIController(jobs_cmds, backend=backend)
an_input = ""
jobs_cmds = handle_job_cmds(jobs_cmds)
bootup()
if not jobs_cmds:
welcome_message()
if first_time_user():
with contextlib.suppress(EOFError):
webbrowser.open("https://docs.openbb.co/cli")
t_controller.print_help()
while ret_code:
if t_controller.queue and len(t_controller.queue) > 0:
if t_controller.queue[0] in ("q", "..", "quit"):
print_goodbye()
break
an_input = t_controller.queue[0]
t_controller.queue = t_controller.queue[1:]
if an_input and an_input.split(" ")[0] in t_controller.CHOICES_COMMANDS:
session.console.print(f"{get_flair_and_username()} / $ {an_input}")
else:
try:
if session.prompt_session and session.settings.USE_PROMPT_TOOLKIT:
if session.settings.TOOLBAR_HINT:
an_input = session.prompt_session.prompt(
f"{get_flair_and_username()} / $ ",
completer=t_controller.completer,
search_ignore_case=True,
bottom_toolbar=HTML(
' help menu '
' return to previous menu '
' exit the program '
' '
"see usage and available options "
),
style=Style.from_dict(
{
"bottom-toolbar": "#ffffff bg:#333333",
}
),
)
else:
an_input = session.prompt_session.prompt(
f"{get_flair_and_username()} / $ ",
completer=t_controller.completer,
search_ignore_case=True,
)
else:
an_input = input(f"{get_flair_and_username()} / $ ")
except (KeyboardInterrupt, EOFError):
print_goodbye()
break
try:
t_controller.queue = t_controller.switch(an_input)
if an_input in ("q", "quit", "..", "exit", "e"):
print_goodbye()
break
if an_input in ("r", "reset") or t_controller.update_success:
reset(t_controller.queue if t_controller.queue else [])
break
except SystemExit:
session.console.print(
f"[red]The command '{an_input}' doesn't exist on the / menu.[/red]\n",
)
similar_cmd = difflib.get_close_matches(
an_input.split(" ")[0] if " " in an_input else an_input,
t_controller.controller_choices,
n=1,
cutoff=0.7,
)
if similar_cmd:
an_input = similar_cmd[0]
if " " in an_input: # pragma: no cover
candidate_input = (
f"{similar_cmd[0]} {' '.join(an_input.split(' ')[1:])}"
)
if candidate_input == an_input:
an_input = ""
t_controller.queue = []
session.console.print("\n")
continue
an_input = candidate_input
session.console.print(f"[green]Replacing by '{an_input}'.[/green]")
t_controller.queue.insert(0, an_input)
def insert_start_slash(cmds: list[str]) -> list[str]:
"""Insert a slash at the beginning of a command sequence."""
if not cmds[0].startswith("/"):
cmds[0] = f"/{cmds[0]}"
if cmds[0].startswith("/home"):
cmds[0] = f"/{cmds[0][5:]}"
return cmds
def run_scripts(
path: Path,
test_mode: bool = False,
verbose: bool = False,
routines_args: list[str] | None = None,
special_arguments: dict[str, str] | None = None,
output: bool = True,
):
"""Run given .openbb scripts.
Parameters
----------
path : Path
The location of the .openbb file.
test_mode : bool
Whether the CLI is in test mode.
verbose : bool
Whether to run tests in verbose mode.
routines_args : list[str] | None
One or multiple inputs to be replaced in the routine and separated by commas.
special_arguments : dict[str, str] | None
Replace `${key=default}` with `value` for every key in the dictionary.
output : bool
Whether to log tests to txt files.
"""
if not path.exists():
session.console.print(f"File '{path}' doesn't exist. Launching base CLI.\n")
if not test_mode:
run_cli()
with path.open() as fp:
raw_lines = [x for x in fp if (not is_reset(x)) and ("#" not in x) and x]
raw_lines = [
raw_line.strip("\n") for raw_line in raw_lines if raw_line.strip("\n")
]
if routines_args:
lines = []
for rawline in raw_lines:
templine = rawline
for i, arg in enumerate(routines_args):
templine = templine.replace(f"$ARGV[{i}]", arg)
lines.append(templine)
elif special_arguments:
lines = []
for line in raw_lines:
new_line = re.sub(
r"\${[^{]+=[^{]+}",
lambda x: replace_dynamic(x, special_arguments),
line,
)
lines.append(new_line)
else:
lines = raw_lines
if test_mode and "exit" not in lines[-1]:
lines.append("exit")
export_folder = ""
if "export" in lines[0]:
export_folder = lines[0].split("export ")[1].rstrip()
lines = lines[1:]
simulate_argv = f"/{'/'.join([line.rstrip() for line in lines])}"
file_cmds = simulate_argv.replace("//", "/home/").split()
file_cmds = insert_start_slash(file_cmds) if file_cmds else file_cmds
file_cmds = (
[f"export {export_folder}{' '.join(file_cmds)}"]
if export_folder
else [" ".join(file_cmds)]
)
if not test_mode or verbose:
run_cli(file_cmds, test_mode=True)
else:
with suppress_stdout():
session.console.print(f"To ensure: {output}")
if output:
timestamp = datetime.now().timestamp()
stamp_str = str(timestamp).replace(".", "")
whole_path = Path(REPOSITORY_DIRECTORY / "integration_test_output")
whole_path.mkdir(parents=True, exist_ok=True)
first_cmd = file_cmds[0].split("/")[1]
with (
open(
whole_path / f"{stamp_str}_{first_cmd}_output.txt", "w"
) as output_file,
contextlib.redirect_stdout(output_file),
):
run_cli(file_cmds, test_mode=True)
else:
run_cli(file_cmds, test_mode=True)
def replace_dynamic(match: re.Match, special_arguments: dict[str, str]) -> str:
"""Replace ${key=default} with value in special_arguments if it exists, else with default.
Parameters
----------
match : re.Match
The match object.
special_arguments : dict[str, str]
The key value pairs to replace in the scripts.
Returns
-------
str
The new string.
"""
cleaned = match[0].replace("{", "").replace("}", "").replace("$", "")
key, default = cleaned.split("=")
dict_value = special_arguments.get(key, default)
if dict_value:
return dict_value
return default
def run_routine(file: str, routines_args: str | None = None):
"""Execute command routine from .openbb file."""
user_routine_path = Path(session.user.preferences.export_directory, "routines")
default_routine_path = ASSETS_DIRECTORY / "routines" / file
if user_routine_path.exists():
run_scripts(
path=user_routine_path,
routines_args=[routines_args] if routines_args else None,
)
elif default_routine_path.exists():
run_scripts(
path=default_routine_path,
routines_args=[routines_args] if routines_args else None,
)
else:
session.console.print(
f"Routine not found, please put your `.openbb` file into : {user_routine_path}."
)
def main(
debug: bool,
dev: bool,
path_list: list[str],
routines_args: list[str] | None = None,
*,
backend: Backend | None = None,
**kwargs,
):
"""Run the CLI with various options.
Parameters
----------
debug : bool
Whether to run the CLI in debug mode.
dev : bool
Points backend towards development environment instead of production.
path_list : list[str]
The paths to run for scripts or to test.
routines_args : list[str] | None
One or multiple inputs to be replaced in the routine and separated by commas.
"""
if debug:
session.settings.DEBUG_MODE = True
if dev:
session.settings.DEV_BACKEND = True
if isinstance(path_list, list) and path_list[0].endswith(".openbb"):
run_routine(
file=path_list[0],
routines_args=",".join(routines_args) if routines_args else None,
)
elif path_list:
argv_cmds = list([" ".join(path_list).replace(" /", "/home/")])
argv_cmds = insert_start_slash(argv_cmds) if argv_cmds else argv_cmds
run_cli(argv_cmds, backend=backend)
else:
run_cli(backend=backend)
def parse_args_and_run():
"""Parse input arguments and run CLI."""
parser = argparse.ArgumentParser(
formatter_class=argparse.ArgumentDefaultsHelpFormatter,
prog="cli",
description="The OpenBB Platform CLI.",
)
parser.add_argument(
"-d",
"--debug",
dest="debug",
action="store_true",
default=False,
help="Runs the CLI in debug mode.",
)
parser.add_argument(
"--dev",
dest="dev",
action="store_true",
default=False,
help="Points backend towards development environment instead of production",
)
parser.add_argument(
"--file",
help="The path or .openbb file to run.",
dest="path",
nargs="+",
default="",
type=str,
)
parser.add_argument(
"-i",
"--input",
help=(
"Select multiple inputs to be replaced in the routine and separated by commas.E.g. GME,AMC,BTC-USD"
),
dest="routine_args",
type=lambda s: [str(item) for item in s.split(",")],
default=None,
)
parser.add_argument(
"-t",
"--test",
action="store_true",
help=(
"Run the CLI in testing mode. Also run this option and '-h' to see testing argument options."
),
)
parser.add_argument(
"-m",
help=argparse.SUPPRESS,
dest="module",
default="",
type=str,
)
parser.add_argument(
"-f",
help=argparse.SUPPRESS,
dest="module_file",
default="",
type=str,
)
parser.add_argument(
"--HistoryManager.hist_file",
help=argparse.SUPPRESS,
dest="module_hist_file",
default="",
type=str,
)
if sys.argv[1:] and "-" not in sys.argv[1][0]:
sys.argv.insert(1, "--file")
ns_parser, unknown = parser.parse_known_args()
if unknown:
if ns_parser.debug:
session.console.print(unknown)
else:
sys.exit(-1)
main(
ns_parser.debug,
ns_parser.dev,
ns_parser.path,
ns_parser.routine_args,
module=ns_parser.module,
module_file=ns_parser.module_file,
module_hist_file=ns_parser.module_hist_file,
)
def launch(
debug: bool = False,
dev: bool = False,
queue: list[str] | None = None,
*,
backend: Backend | None = None,
) -> None:
"""Launch CLI, optionally with a non-default ``Backend``."""
if queue:
main(debug, dev, queue, module="", backend=backend)
else:
parse_args_and_run()
if __name__ == "__main__":
parse_args_and_run()