summaryrefslogtreecommitdiffstats
diff options
context:
space:
mode:
authorDiogo Sousa <montezdesousa@gmail.com>2024-04-30 13:54:32 +0100
committerDiogo Sousa <montezdesousa@gmail.com>2024-04-30 13:54:32 +0100
commitdc5bccdc26594246b3739a041ced1a744ca83282 (patch)
tree43ca06c1479e1c3a966e760da55e0b91726ef409
parent7eff3f01a54c2b1b511223d1d181ff47065e8e8e (diff)
fix: avoid Session()
-rw-r--r--cli/openbb_cli/controllers/cli_controller.py77
-rw-r--r--cli/openbb_cli/controllers/feature_flags_controller.py86
2 files changed, 79 insertions, 84 deletions
diff --git a/cli/openbb_cli/controllers/cli_controller.py b/cli/openbb_cli/controllers/cli_controller.py
index f6e0998fc2a..8c79ae6f842 100644
--- a/cli/openbb_cli/controllers/cli_controller.py
+++ b/cli/openbb_cli/controllers/cli_controller.py
@@ -66,6 +66,7 @@ DATA_PROCESSING_ROUTERS = ["technical", "quantitative", "econometrics"]
logger = logging.getLogger(__name__)
env_file = str(ENV_FILE_SETTINGS)
+session = Session()
if is_installer():
# Necessary for installer so that it can locate the correct certificates for
@@ -163,11 +164,9 @@ class CLIController(BaseController):
def update_runtime_choices(self):
"""Update runtime choices."""
- routines_directory = Path(
- Session().user.preferences.export_directory, "routines"
- )
+ routines_directory = Path(session.user.preferences.export_directory, "routines")
- if Session().prompt_session and Session().settings.USE_PROMPT_TOOLKIT:
+ if session.prompt_session and session.settings.USE_PROMPT_TOOLKIT:
# choices: dict = self.choices_default
choices: dict = {c: {} for c in self.controller_choices} # type: ignore
choices["hold"] = {c: None for c in ["on", "off", "-s", "--sameaxis"]}
@@ -241,8 +240,8 @@ class CLIController(BaseController):
.get("description")
) or ""
mt.add_menu(
- key_menu=router,
- menu_description=menu_description.split(".")[0].lower(),
+ name=router,
+ description=menu_description.split(".")[0].lower(),
)
else:
mt.add_cmd(router)
@@ -259,8 +258,8 @@ class CLIController(BaseController):
.get("description")
) or ""
mt.add_menu(
- key_menu=router,
- menu_description=menu_description.split(".")[0].lower(),
+ name=router,
+ description=menu_description.split(".")[0].lower(),
)
else:
mt.add_cmd(router)
@@ -276,8 +275,8 @@ class CLIController(BaseController):
.get("description")
) or ""
mt.add_menu(
- key_menu=router,
- menu_description=menu_description.split(".")[0].lower(),
+ name=router,
+ description=menu_description.split(".")[0].lower(),
)
else:
mt.add_cmd(router)
@@ -285,7 +284,7 @@ class CLIController(BaseController):
mt.add_raw("\n cached results (OBBjects)\n")
mt.add_cmd("results")
- Session().console.print(text=mt.menu_text, menu="Home")
+ session.console.print(text=mt.menu_text, menu="Home")
self.update_runtime_choices()
def parse_input(self, an_input: str) -> List:
@@ -314,7 +313,7 @@ class CLIController(BaseController):
other_args += self.queue
if not other_args:
- Session().console.print(
+ session.console.print(
"[info]Provide a path to the routine you wish to execute. For an example, please use "
"`exe --example` and for documentation and to learn how create your own script "
"type `about exe`.\n[/info]"
@@ -369,7 +368,7 @@ class CLIController(BaseController):
if ns_parser:
if ns_parser.example:
routine_path = ASSETS_DIRECTORY / "routines" / "routine_example.openbb"
- Session().console.print(
+ session.console.print(
"[info]Executing an example, please type `about exe` "
"to learn how to create your own script.[/info]\n"
)
@@ -389,14 +388,12 @@ class CLIController(BaseController):
final_url = f"{url}?raw=true"
response = requests.get(final_url, timeout=10)
if response.status_code != 200:
- Session().console.print(
+ session.console.print(
"[red]Could not find the requested script.[/red]"
)
return
routine_text = response.json()["script"]
- file_path = Path(
- Session().user.preferences.export_directory, "routines"
- )
+ file_path = Path(session.user.preferences.export_directory, "routines")
routine_path = file_path / file_name
with open(routine_path, "w") as file:
file.write(routine_text)
@@ -441,7 +438,7 @@ class CLIController(BaseController):
# issue in parsing the routine and therefore we don't want to feed it
# to the terminal
if err:
- Session().console.print(err)
+ session.console.print(err)
return
self.queue = [
@@ -466,12 +463,12 @@ class CLIController(BaseController):
# Check if the directory exists
if os.path.isdir(export_path):
- Session().console.print(
+ session.console.print(
f"Export data to be saved in the selected folder: '{export_path}'"
)
else:
os.makedirs(export_path)
- Session().console.print(
+ session.console.print(
f"[green]Folder '{export_path}' successfully created.[/green]"
)
self.queue = self.queue[1:]
@@ -485,7 +482,7 @@ class CLIController(BaseController):
df, show_index=True, index_name="stack index", title="OBBject Results"
)
else:
- Session().console.print("[info]No results found.[/info]")
+ session.console.print("[info]No results found.[/info]")
def handle_job_cmds(jobs_cmds: Optional[List[str]]) -> Optional[List[str]]:
@@ -513,12 +510,12 @@ def handle_job_cmds(jobs_cmds: Optional[List[str]]) -> Optional[List[str]]:
# Check if the directory exists
if os.path.isdir(export_path):
- Session().console.print(
+ session.console.print(
f"Export data to be saved in the selected folder: '{export_path}'"
)
else:
os.makedirs(export_path)
- Session().console.print(
+ session.console.print(
f"[green]Folder '{export_path}' successfully created.[/green]"
)
return jobs_cmds
@@ -560,16 +557,16 @@ def run_cli(jobs_cmds: Optional[List[str]] = None, test_mode=False):
# Print the current location because this was an instruction and we want user to know what was the action
if an_input and an_input.split(" ")[0] in t_controller.CHOICES_COMMANDS:
- Session().console.print(f"{get_flair_and_username()} / $ {an_input}")
+ session.console.print(f"{get_flair_and_username()} / $ {an_input}")
# Get input command from user
else:
try:
# Get input from user using auto-completion
- if Session().prompt_session and Session().settings.USE_PROMPT_TOOLKIT:
+ if session.prompt_session and session.settings.USE_PROMPT_TOOLKIT:
# Check if toolbar hint was enabled
- if Session().settings.TOOLBAR_HINT:
- an_input = Session().prompt_session.prompt( # type: ignore[union-attr]
+ if session.settings.TOOLBAR_HINT:
+ an_input = session.prompt_session.prompt( # type: ignore[union-attr]
f"{get_flair_and_username()} / $ ",
completer=t_controller.completer,
search_ignore_case=True,
@@ -588,7 +585,7 @@ def run_cli(jobs_cmds: Optional[List[str]] = None, test_mode=False):
),
)
else:
- an_input = Session().prompt_session.prompt( # type: ignore[union-attr]
+ an_input = session.prompt_session.prompt( # type: ignore[union-attr]
f"{get_flair_and_username()} / $ ",
completer=t_controller.completer,
search_ignore_case=True,
@@ -620,7 +617,7 @@ def run_cli(jobs_cmds: Optional[List[str]] = None, test_mode=False):
"The command '%s' doesn't exist on the / menu.",
an_input,
)
- Session().console.print(
+ session.console.print(
f"[red]The command '{an_input}' doesn't exist on the / menu.[/red]\n",
)
similar_cmd = difflib.get_close_matches(
@@ -638,11 +635,11 @@ def run_cli(jobs_cmds: Optional[List[str]] = None, test_mode=False):
if candidate_input == an_input:
an_input = ""
t_controller.queue = []
- Session().console.print("\n")
+ session.console.print("\n")
continue
an_input = candidate_input
- Session().console.print(f"[green]Replacing by '{an_input}'.[/green]")
+ session.console.print(f"[green]Replacing by '{an_input}'.[/green]")
t_controller.queue.insert(0, an_input)
@@ -682,7 +679,7 @@ def run_scripts(
Whether to log tests to txt files
"""
if not path.exists():
- Session().console.print(f"File '{path}' doesn't exist. Launching base CLI.\n")
+ session.console.print(f"File '{path}' doesn't exist. Launching base CLI.\n")
if not test_mode:
run_cli()
@@ -736,7 +733,7 @@ def run_scripts(
run_cli(file_cmds, test_mode=True)
else:
with suppress_stdout():
- Session().console.print(f"To ensure: {output}")
+ session.console.print(f"To ensure: {output}")
if output:
timestamp = datetime.now().timestamp()
stamp_str = str(timestamp).replace(".", "")
@@ -777,7 +774,7 @@ def replace_dynamic(match: re.Match, special_arguments: Dict[str, str]) -> str:
def run_routine(file: str, routines_args=Optional[str]):
"""Execute command routine from .openbb file."""
- user_routine_path = Path(Session().user.preferences.export_directory, "routines")
+ user_routine_path = Path(session.user.preferences.export_directory, "routines")
default_routine_path = ASSETS_DIRECTORY / "routines" / file
if user_routine_path.exists():
@@ -785,7 +782,7 @@ def run_routine(file: str, routines_args=Optional[str]):
elif default_routine_path.exists():
run_scripts(path=default_routine_path, routines_args=routines_args)
else:
- Session().console.print(
+ session.console.print(
f"Routine not found, please put your `.openbb` file into : {user_routine_path}."
)
@@ -819,12 +816,12 @@ def main(
E.g. GME,AMC,BTC-USD
"""
if debug:
- Session().settings.DEBUG_MODE = True
+ session.settings.DEBUG_MODE = True
if dev:
- Session().settings.DEV_BACKEND = True
- Session().settings.BASE_URL = "https://payments.openbb.dev/"
- Session().settings.HUB_URL = "https://my.openbb.dev"
+ session.settings.DEV_BACKEND = True
+ session.settings.BASE_URL = "https://payments.openbb.dev/"
+ session.settings.HUB_URL = "https://my.openbb.dev"
if isinstance(path_list, list) and path_list[0].endswith(".openbb"):
run_routine(file=path_list[0], routines_args=routines_args)
@@ -917,7 +914,7 @@ def parse_args_and_run():
# Use -d flag if you want to see the unknown args.
if unknown:
if ns_parser.debug:
- Session().console.print(unknown)
+ session.console.print(unknown)
else:
sys.exit(-1)
diff --git a/cli/openbb_cli/controllers/feature_flags_controller.py b/cli/openbb_cli/controllers/feature_flags_controller.py
index 860cd98a819..456ea20d5ba 100644
--- a/cli/openbb_cli/controllers/feature_flags_controller.py
+++ b/cli/openbb_cli/controllers/feature_flags_controller.py
@@ -13,6 +13,8 @@ from openbb_cli.controllers.base_controller import BaseController
from openbb_cli.controllers.utils import all_timezones, is_timezone_valid
from openbb_cli.session import Session
+session = Session()
+
class FeatureFlagsController(BaseController):
"""Feature Flags Controller class."""
@@ -46,13 +48,13 @@ class FeatureFlagsController(BaseController):
"""Initialize the Constructor."""
super().__init__(queue)
- if Session().prompt_session and Session().settings.USE_PROMPT_TOOLKIT:
+ if session.prompt_session and session.settings.USE_PROMPT_TOOLKIT:
choices: dict = self.choices_default
self.completer = NestedCompleter.from_nested_dict(choices)
def print_help(self):
"""Print help."""
- settings = Session().settings
+ settings = session.settings
mt = MenuText("settings/")
mt.add_info("_info_")
@@ -76,63 +78,61 @@ class FeatureFlagsController(BaseController):
mt.add_cmd("n_rows")
mt.add_cmd("n_cols")
- Session().console.print(text=mt.menu_text, menu="Feature Flags")
+ session.console.print(text=mt.menu_text, menu="Feature Flags")
def call_overwrite(self, _):
"""Process overwrite command."""
- Session().settings.set_item(
- "FILE_OVERWRITE", not Session().settings.FILE_OVERWRITE
- )
+ session.settings.set_item("FILE_OVERWRITE", not session.settings.FILE_OVERWRITE)
def call_version(self, _):
"""Process version command."""
- Session().settings.SHOW_VERSION = not Session().settings.SHOW_VERSION
+ session.settings.SHOW_VERSION = not session.settings.SHOW_VERSION
def call_interactive(self, _):
"""Process interactive command."""
- Session().settings.set_item(
- "USE_INTERACTIVE_DF", not Session().settings.USE_INTERACTIVE_DF
+ session.settings.set_item(
+ "USE_INTERACTIVE_DF", not session.settings.USE_INTERACTIVE_DF
)
def call_cls(self, _):
"""Process cls command."""
- Session().settings.set_item(
- "USE_CLEAR_AFTER_CMD", not Session().settings.USE_CLEAR_AFTER_CMD
+ session.settings.set_item(
+ "USE_CLEAR_AFTER_CMD", not session.settings.USE_CLEAR_AFTER_CMD
)
def call_promptkit(self, _):
"""Process promptkit command."""
- Session().settings.set_item(
- "USE_PROMPT_TOOLKIT", not Session().settings.USE_PROMPT_TOOLKIT
+ session.settings.set_item(
+ "USE_PROMPT_TOOLKIT", not session.settings.USE_PROMPT_TOOLKIT
)
def call_exithelp(self, _):
"""Process exithelp command."""
- Session().settings.set_item(
- "ENABLE_EXIT_AUTO_HELP", not Session().settings.ENABLE_EXIT_AUTO_HELP
+ session.settings.set_item(
+ "ENABLE_EXIT_AUTO_HELP", not session.settings.ENABLE_EXIT_AUTO_HELP
)
def call_rcontext(self, _):
"""Process rcontext command."""
- Session().settings.set_item(
- "REMEMBER_CONTEXTS", not Session().settings.REMEMBER_CONTEXTS
+ session.settings.set_item(
+ "REMEMBER_CONTEXTS", not session.settings.REMEMBER_CONTEXTS
)
def call_dt(self, _):
"""Process dt command."""
- Session().settings.set_item("USE_DATETIME", not Session().settings.USE_DATETIME)
+ session.settings.set_item("USE_DATETIME", not session.settings.USE_DATETIME)
def call_richpanel(self, _):
"""Process richpanel command."""
- Session().settings.set_item(
- "ENABLE_RICH_PANEL", not Session().settings.ENABLE_RICH_PANEL
+ session.settings.set_item(
+ "ENABLE_RICH_PANEL", not session.settings.ENABLE_RICH_PANEL
)
def call_tbhint(self, _):
"""Process tbhint command."""
- if Session().settings.TOOLBAR_HINT:
- Session().console.print("Will take effect when running CLI again.")
- Session().settings.set_item("TOOLBAR_HINT", not Session().settings.TOOLBAR_HINT)
+ if session.settings.TOOLBAR_HINT:
+ session.console.print("Will take effect when running CLI again.")
+ session.settings.set_item("TOOLBAR_HINT", not session.settings.TOOLBAR_HINT)
def call_console_style(self, other_args: List[str]) -> None:
"""Process cosole_style command."""
@@ -148,16 +148,16 @@ class FeatureFlagsController(BaseController):
dest="style",
action="store",
required=False,
- choices=Session().style.available_styles,
+ choices=session.style.available_styles,
)
ns_parser = self.parse_simple_args(parser, other_args)
if ns_parser and ns_parser.style:
- Session().style.apply(ns_parser.style)
- Session().settings.set_item("RICH_STYLE", ns_parser.style)
+ session.style.apply(ns_parser.style)
+ session.settings.set_item("RICH_STYLE", ns_parser.style)
elif not other_args:
- Session().console.print(
- f"Current console style: {Session().settings.RICH_STYLE}"
+ session.console.print(
+ f"Current console style: {session.settings.RICH_STYLE}"
)
def call_flair(self, other_args: List[str]) -> None:
@@ -179,9 +179,9 @@ class FeatureFlagsController(BaseController):
ns_parser = self.parse_simple_args(parser, other_args)
if ns_parser and ns_parser.flair:
- Session().settings.set_item("FLAIR", ns_parser.flair)
+ session.settings.set_item("FLAIR", ns_parser.flair)
elif not other_args:
- Session().console.print(f"Current flair: {Session().settings.FLAIR}")
+ session.console.print(f"Current flair: {session.settings.FLAIR}")
def call_timezone(self, other_args: List[str]) -> None:
"""Process timezone command."""
@@ -203,16 +203,16 @@ class FeatureFlagsController(BaseController):
if ns_parser and ns_parser.timezone:
if is_timezone_valid(ns_parser.timezone):
- Session().settings.set_item("TIMEZONE", ns_parser.timezone)
+ session.settings.set_item("TIMEZONE", ns_parser.timezone)
else:
- Session().console.print(
+ session.console.print(
"Invalid timezone. Please enter a valid timezone."
)
- Session().console.print(
+ session.console.print(
f"Available timezones are: {', '.join(all_timezones)}"
)
elif not other_args:
- Session().console.print(f"Current timezone: {Session().settings.TIMEZONE}")
+ session.console.print(f"Current timezone: {session.settings.TIMEZONE}")
def call_language(self, other_args: List[str]) -> None:
"""Process language command."""
@@ -233,12 +233,10 @@ class FeatureFlagsController(BaseController):
ns_parser = self.parse_simple_args(parser, other_args)
if ns_parser and ns_parser.language:
- Session().settings.set_item("USE_LANGUAGE", ns_parser.language)
+ session.settings.set_item("USE_LANGUAGE", ns_parser.language)
elif not other_args:
- Session().console.print(
- f"Current language: {Session().settings.USE_LANGUAGE}"
- )
+ session.console.print(f"Current language: {session.settings.USE_LANGUAGE}")
def call_n_rows(self, other_args: List[str]) -> None:
"""Process n_rows command."""
@@ -259,11 +257,11 @@ class FeatureFlagsController(BaseController):
ns_parser = self.parse_simple_args(parser, other_args)
if ns_parser and ns_parser.rows:
- Session().settings.set_item("ALLOWED_NUMBER_OF_ROWS", ns_parser.rows)
+ session.settings.set_item("ALLOWED_NUMBER_OF_ROWS", ns_parser.rows)
elif not other_args:
- Session().console.print(
- f"Current number of rows: {Session().settings.ALLOWED_NUMBER_OF_ROWS}"
+ session.console.print(
+ f"Current number of rows: {session.settings.ALLOWED_NUMBER_OF_ROWS}"
)
def call_n_cols(self, other_args: List[str]) -> None:
@@ -285,9 +283,9 @@ class FeatureFlagsController(BaseController):
ns_parser = self.parse_simple_args(parser, other_args)
if ns_parser and ns_parser.columns:
- Session().settings.set_item("ALLOWED_NUMBER_OF_COLUMNS", ns_parser.columns)
+ session.settings.set_item("ALLOWED_NUMBER_OF_COLUMNS", ns_parser.columns)
elif not other_args:
- Session().console.print(
- f"Current number of columns: {Session().settings.ALLOWED_NUMBER_OF_COLUMNS}"
+ session.console.print(
+ f"Current number of columns: {session.settings.ALLOWED_NUMBER_OF_COLUMNS}"
)