Skip to content

Commit

Permalink
Merge branch 'develop' into feature/cli-unit-tests
Browse files Browse the repository at this point in the history
  • Loading branch information
@montezdesousa
montezdesousa committed May 14, 2024
2 parents 590bc10 + c2f5f7a commit a9318c0
Show file tree
Hide file tree
Showing 9 changed files with 221 additions and 207 deletions.
6 changes: 4 additions & 2 deletions cli/openbb_cli/argparse_translator/obbject_registry.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down
9 changes: 5 additions & 4 deletions cli/openbb_cli/config/menu_text.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -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"

Expand Down
214 changes: 114 additions & 100 deletions cli/openbb_cli/controllers/base_controller.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -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."""
Expand Down
15 changes: 10 additions & 5 deletions cli/openbb_cli/controllers/base_platform_controller.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -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]"
Expand All @@ -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)
Expand Down Expand Up @@ -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)

Expand Down
Loading

0 comments on commit a9318c0

Please sign in to comment.