Merge branch 'develop' into feature/docs-cli

cli/openbb_cli/argparse_translator/argparse_translator.py

@@ -42,7 +42,7 @@ class CustomArgument(BaseModel):
     action: Literal["store_true", "store"]
     help: str
     nargs: Optional[Literal["+"]]
-    choices: Optional[Any]
+    choices: Optional[Tuple]
 
     @model_validator(mode="after")  # type: ignore
     @classmethod
@@ -117,7 +117,7 @@ def _get_nargs(self, type_: type) -> Optional[Union[int, str]]:
             return "+"
         return None
 
-    def _get_choices(self, type_: str) -> Tuple:
+    def _get_choices(self, type_: str, custom_choices: Any) -> Tuple:
         """Get the choices for the given type."""
         type_ = self._make_type_parsable(type_)  # type: ignore
         type_origin = get_origin(type_)
@@ -126,14 +126,12 @@ def _get_choices(self, type_: str) -> Tuple:
 
         if type_origin is Literal:
             choices = get_args(type_)
-            # param_type = type(choices[0])
 
         if type_origin is list:
             type_ = get_args(type_)[0]
 
             if get_origin(type_) is Literal:
                 choices = get_args(type_)
-                # param_type = type(choices[0])
 
         if type_origin is Union and type(None) in get_args(type_):
             # remove NoneType from the args
@@ -145,7 +143,9 @@ def _get_choices(self, type_: str) -> Tuple:
 
             if get_origin(type_) is Literal:
                 choices = get_args(type_)
-                # param_type = type(choices[0])
+
+        if custom_choices:
+            return tuple(custom_choices)
 
         return choices
 
@@ -174,7 +174,9 @@ def build_custom_groups(self):
                             action="store" if type_ != bool else "store_true",
                             help=arg["description"],
                             nargs=self._get_nargs(type_),  # type: ignore
-                            choices=self._get_choices(arg["type"]),
+                            choices=self._get_choices(
+                                arg["type"], custom_choices=arg["choices"]
+                            ),
                         )
                     )
 

cli/openbb_cli/argparse_translator/obbject_registry.py

@@ -20,8 +20,10 @@ def _contains_obbject(uuid: str, obbjects: List[OBBject]) -> bool:
 
     def register(self, obbject: OBBject) -> bool:
         """Designed to add an OBBject instance to the registry."""
-        if isinstance(obbject, OBBject) and not self._contains_obbject(
-            obbject.id, self._obbjects
+        if (
+            isinstance(obbject, OBBject)
+            and not self._contains_obbject(obbject.id, self._obbjects)
+            and obbject.results
         ):
             self._obbjects.append(obbject)
             return True

cli/openbb_cli/config/menu_text.py

@@ -99,17 +99,18 @@ def _format_cmd_description(
             else description
         )
 
-    def add_raw(self, text: str):
+    def add_raw(self, text: str, left_spacing: bool = False):
         """Append raw text (without translation)."""
-        self.menu_text += text
+        if left_spacing:
+            self.menu_text += f"{self.SECTION_SPACING * ' '}{text}\n"
+        else:
+            self.menu_text += text
 
     def add_section(
         self, text: str, description: str = "", leading_new_line: bool = False
     ):
         """Append raw text (without translation)."""
         spacing = (self.CMD_NAME_LENGTH - len(text) + self.SECTION_SPACING) * " "
-        left_spacing = self.SECTION_SPACING * " "
-        text = f"{left_spacing}{text}"
         if description:
             text = f"{text}{spacing}{description}\n"
 

cli/openbb_cli/controllers/base_controller.py

@@ -479,123 +479,137 @@ def call_record(self, other_args) -> None:
                 "\n[yellow]Remember to run 'stop' command when you are done!\n[/yellow]"
             )
 
-    def call_stop(self, _) -> None:
+    def call_stop(self, other_args) -> None:
         """Process stop command."""
-        global RECORD_SESSION  # noqa: PLW0603
-        global SESSION_RECORDED  # noqa: PLW0603
+        parser = argparse.ArgumentParser(
+            add_help=False,
+            formatter_class=argparse.ArgumentDefaultsHelpFormatter,
+            prog="stop",
+            description="Stop recording session into .openbb routine file",
+        )
+        # This is only for auto-completion purposes
+        _ = self.parse_simple_args(parser, other_args)
 
-        if not RECORD_SESSION:
-            session.console.print(
-                "[red]There is no session being recorded. Start one using the command 'record'[/red]\n"
-            )
-        elif len(SESSION_RECORDED) < 5:
-            session.console.print(
-                "[red]Run at least 4 commands before stopping recording a session.[/red]\n"
-            )
-        else:
-            current_user = session.user
+        if "-h" not in other_args and "--help" not in other_args:
+            global RECORD_SESSION  # noqa: PLW0603
+            global SESSION_RECORDED  # noqa: PLW0603
 
-            # Check if the user just wants to store routine locally
-            # This works regardless of whether they are logged in or not
-            if RECORD_SESSION_LOCAL_ONLY:
-                # Whitespaces are replaced by underscores and an .openbb extension is added
-                title_for_local_storage = (
-                    SESSION_RECORDED_NAME.replace(" ", "_") + ".openbb"
+            if not RECORD_SESSION:
+                session.console.print(
+                    "[red]There is no session being recorded. Start one using the command 'record'[/red]\n"
                 )
-
-                routine_file = os.path.join(
-                    f"{current_user.preferences.export_directory}/routines",
-                    title_for_local_storage,
+            elif len(SESSION_RECORDED) < 5:
+                session.console.print(
+                    "[red]Run at least 4 commands before stopping recording a session.[/red]\n"
                 )
+            else:
+                current_user = session.user
+
+                # Check if the user just wants to store routine locally
+                # This works regardless of whether they are logged in or not
+                if RECORD_SESSION_LOCAL_ONLY:
+                    # Whitespaces are replaced by underscores and an .openbb extension is added
+                    title_for_local_storage = (
+                        SESSION_RECORDED_NAME.replace(" ", "_") + ".openbb"
+                    )
 
-                # If file already exists, add a timestamp to the name
-                if os.path.isfile(routine_file):
-                    i = session.console.input(
-                        "A local routine with the same name already exists, "
-                        "do you want to override it? (y/n): "
+                    routine_file = os.path.join(
+                        f"{current_user.preferences.export_directory}/routines",
+                        title_for_local_storage,
                     )
-                    session.console.print("")
-                    while i.lower() not in ["y", "yes", "n", "no"]:
-                        i = session.console.input("Select 'y' or 'n' to proceed: ")
-                        session.console.print("")
 
-                    if i.lower() in ["n", "no"]:
-                        new_name = (
-                            datetime.now().strftime("%Y%m%d_%H%M%S_")
-                            + title_for_local_storage
-                        )
-                        routine_file = os.path.join(
-                            current_user.preferences.export_directory,
-                            "routines",
-                            new_name,
-                        )
-                        session.console.print(
-                            f"[yellow]The routine name has been updated to '{new_name}'[/yellow]\n"
+                    # If file already exists, add a timestamp to the name
+                    if os.path.isfile(routine_file):
+                        i = session.console.input(
+                            "A local routine with the same name already exists, "
+                            "do you want to override it? (y/n): "
                         )
+                        session.console.print("")
+                        while i.lower() not in ["y", "yes", "n", "no"]:
+                            i = session.console.input("Select 'y' or 'n' to proceed: ")
+                            session.console.print("")
+
+                        if i.lower() in ["n", "no"]:
+                            new_name = (
+                                datetime.now().strftime("%Y%m%d_%H%M%S_")
+                                + title_for_local_storage
+                            )
+                            routine_file = os.path.join(
+                                current_user.preferences.export_directory,
+                                "routines",
+                                new_name,
+                            )
+                            session.console.print(
+                                f"[yellow]The routine name has been updated to '{new_name}'[/yellow]\n"
+                            )
 
-                # Writing to file
-                Path(os.path.dirname(routine_file)).mkdir(parents=True, exist_ok=True)
-
-                with open(routine_file, "w") as file1:
-                    lines = ["# OpenBB Platform CLI - Routine", "\n"]
-
-                    username = getattr(
-                        session.user.profile.hub_session, "username", "local"
+                    # Writing to file
+                    Path(os.path.dirname(routine_file)).mkdir(
+                        parents=True, exist_ok=True
                     )
 
-                    lines += [f"# Author: {username}", "\n\n"] if username else ["\n"]
-                    lines += [
-                        f"# Title: {SESSION_RECORDED_NAME}",
-                        "\n",
-                        f"# Tags: {SESSION_RECORDED_TAGS}",
-                        "\n\n",
-                        f"# Description: {SESSION_RECORDED_DESCRIPTION}",
-                        "\n\n",
-                    ]
-                    lines += [c + "\n" for c in SESSION_RECORDED[:-1]]
-                    # Writing data to a file
-                    file1.writelines(lines)
+                    with open(routine_file, "w") as file1:
+                        lines = ["# OpenBB Platform CLI - Routine", "\n"]
 
-                session.console.print(
-                    f"[green]Your routine has been recorded and saved here: {routine_file}[/green]\n"
-                )
+                        username = getattr(
+                            session.user.profile.hub_session, "username", "local"
+                        )
 
-            # If user doesn't specify they want to store routine locally
-            # Confirm that the user is logged in
-            elif not session.is_local():
-                routine = "\n".join(SESSION_RECORDED[:-1])
-                hub_session = current_user.profile.hub_session
+                        lines += (
+                            [f"# Author: {username}", "\n\n"] if username else ["\n"]
+                        )
+                        lines += [
+                            f"# Title: {SESSION_RECORDED_NAME}",
+                            "\n",
+                            f"# Tags: {SESSION_RECORDED_TAGS}",
+                            "\n\n",
+                            f"# Description: {SESSION_RECORDED_DESCRIPTION}",
+                            "\n\n",
+                        ]
+                        lines += [c + "\n" for c in SESSION_RECORDED[:-1]]
+                        # Writing data to a file
+                        file1.writelines(lines)
 
-                if routine is not None:
-                    auth_header = (
-                        f"{hub_session.token_type} {hub_session.access_token.get_secret_value()}"
-                        if hub_session
-                        else None
+                    session.console.print(
+                        f"[green]Your routine has been recorded and saved here: {routine_file}[/green]\n"
                     )
-                    kwargs = {
-                        "auth_header": auth_header,
-                        "name": SESSION_RECORDED_NAME,
-                        "description": SESSION_RECORDED_DESCRIPTION,
-                        "routine": routine,
-                        "tags": SESSION_RECORDED_TAGS,
-                        "public": SESSION_RECORDED_PUBLIC,
-                    }
-                    response = upload_routine(**kwargs)  # type: ignore
-                    if response is not None and response.status_code == 409:
-                        i = session.console.input(
-                            "A routine with the same name already exists, "
-                            "do you want to replace it? (y/n): "
-                        )
-                        session.console.print("")
-                        if i.lower() in ["y", "yes"]:
-                            kwargs["override"] = True  # type: ignore
-                            response = upload_routine(**kwargs)  # type: ignore
-                        else:
-                            session.console.print("[info]Aborted.[/info]")
 
-            # Clear session to be recorded again
-            RECORD_SESSION = False
-            SESSION_RECORDED = list()
+                # If user doesn't specify they want to store routine locally
+                # Confirm that the user is logged in
+                elif not session.is_local():
+                    routine = "\n".join(SESSION_RECORDED[:-1])
+                    hub_session = current_user.profile.hub_session
+
+                    if routine is not None:
+                        auth_header = (
+                            f"{hub_session.token_type} {hub_session.access_token.get_secret_value()}"
+                            if hub_session
+                            else None
+                        )
+                        kwargs = {
+                            "auth_header": auth_header,
+                            "name": SESSION_RECORDED_NAME,
+                            "description": SESSION_RECORDED_DESCRIPTION,
+                            "routine": routine,
+                            "tags": SESSION_RECORDED_TAGS,
+                            "public": SESSION_RECORDED_PUBLIC,
+                        }
+                        response = upload_routine(**kwargs)  # type: ignore
+                        if response is not None and response.status_code == 409:
+                            i = session.console.input(
+                                "A routine with the same name already exists, "
+                                "do you want to replace it? (y/n): "
+                            )
+                            session.console.print("")
+                            if i.lower() in ["y", "yes"]:
+                                kwargs["override"] = True  # type: ignore
+                                response = upload_routine(**kwargs)  # type: ignore
+                            else:
+                                session.console.print("[info]Aborted.[/info]")
+
+                # Clear session to be recorded again
+                RECORD_SESSION = False
+                SESSION_RECORDED = list()
 
     def call_whoami(self, other_args: List[str]) -> None:
         """Process whoami command."""

cli/openbb_cli/controllers/base_platform_controller.py

@@ -159,8 +159,8 @@ def method(self, other_args: List[str], translator=translator):
 
                     if obbject:
 
-                        if isinstance(obbject, OBBject) and obbject.results:
-                            if session.max_obbjects_exceeded():
+                        if isinstance(obbject, OBBject):
+                            if session.max_obbjects_exceeded() and obbject.results:
                                 session.obbject_registry.remove()
                                 session.console.print(
                                     "[yellow]Maximum number of OBBjects reached. The oldest entry was removed.[yellow]"
@@ -181,7 +181,9 @@ def method(self, other_args: List[str], translator=translator):
                                 session.settings.SHOW_MSG_OBBJECT_REGISTRY
                                 and register_result
                             ):
-                                session.console.print("Added OBBject to registry.")
+                                session.console.print(
+                                    "Added `OBBject` to cached results."
+                                )
 
                             # making the dataframe available
                             # either for printing or exporting (or both)
@@ -326,11 +328,14 @@ def print_help(self):
                 )
 
         if session.obbject_registry.obbjects:
-            mt.add_section("Cached Results:\n", leading_new_line=True)
+            mt.add_info("\nCached Results")
             for key, value in list(session.obbject_registry.all.items())[
                 : session.settings.N_TO_DISPLAY_OBBJECT_REGISTRY
             ]:
-                mt.add_raw(f"\tOBB{key}: {value['command']}\n")
+                mt.add_raw(
+                    f"[yellow]OBB{key}[/yellow]: {value['command']}",
+                    left_spacing=True,
+                )
 
         session.console.print(text=mt.menu_text, menu=self.PATH)