diff options
author | hjoaquim <h.joaquim@campus.fct.unl.pt> | 2024-04-23 11:15:22 +0100 |
---|---|---|
committer | hjoaquim <h.joaquim@campus.fct.unl.pt> | 2024-04-23 11:15:22 +0100 |
commit | f3b82d6e1d81ea2871f77231a541f1ba448a05c3 (patch) | |
tree | dd633a005e3257fda80f5eb59c99873950b701ba | |
parent | 0c2a92e86efa507dfdf393dec2d00c528f7f09b1 (diff) |
improved obbject registry using a stack
-rw-r--r-- | argparse_translator/obbject_registry.py | 85 | ||||
-rw-r--r-- | openbb_terminal/base_platform_controller.py | 8 | ||||
-rw-r--r-- | openbb_terminal/terminal_controller.py | 2 |
3 files changed, 45 insertions, 50 deletions
diff --git a/argparse_translator/obbject_registry.py b/argparse_translator/obbject_registry.py index 6014ae05ceb..cb1643747b3 100644 --- a/argparse_translator/obbject_registry.py +++ b/argparse_translator/obbject_registry.py @@ -1,3 +1,5 @@ +"""Registry for OBBjects.""" + import json from typing import Dict, List @@ -7,43 +9,31 @@ from openbb_core.app.model.obbject import OBBject class Registry(metaclass=SingletonMeta): - obbject_instances: List[OBBject] = [] - ids: List[str] = [] + obbjects: List[OBBject] = [] - @classmethod - def register(cls, obbject: OBBject): - if isinstance(obbject, OBBject): - for obbject_ in cls.obbject_instances: - if obbject_.id == obbject.id: - return - cls.obbject_instances.append(obbject) - cls.ids.append(obbject.id) - - @classmethod - def remove(cls, id_: str): - for obbject in cls.obbject_instances: - if obbject.id == id_: - cls.obbject_instances.remove(obbject) - cls.ids.remove(id_) + @staticmethod + def _contains_obbject(uuid: str, obbjects: List[OBBject]) -> bool: + """Check if obbject with uuid is in the registry.""" + return any(obbject.id == uuid for obbject in obbjects) @classmethod - def get(cls, id_: str) -> OBBject: - for obbject in cls.obbject_instances: - if obbject.id == id_: - return obbject - raise ValueError(f"OBBject with id {id_} not found") + def register(cls, obbject: OBBject): + """Designed to add an OBBject instance to the registry.""" + if isinstance(obbject, OBBject) and not cls._contains_obbject( + obbject.id, cls.obbjects + ): + cls.obbjects.append(obbject) @classmethod - def pop(cls, id_: str) -> OBBject: - for obbject in cls.obbject_instances: - if obbject.id == id_: - cls.obbject_instances.remove(obbject) - cls.ids.remove(id_) - return obbject - raise ValueError(f"OBBject with id {id_} not found") + def get(cls, idx: int) -> OBBject: + """Return the obbject at index idx.""" + # the list should work as a stack + # i.e., the last element needs to be accessed by idx=0 and so on + reversed_list = list(reversed(cls.obbjects)) + return reversed_list[idx] @property - def all(self) -> Dict[str, Dict]: + def all(self) -> Dict[int, Dict]: """Return all obbjects in the registry""" def _handle_standard_params(obbject: OBBject) -> str: @@ -52,28 +42,35 @@ class Registry(metaclass=SingletonMeta): std_params = obbject._standard_params # pylint: disable=protected-access if hasattr(std_params, "__dict__"): standard_params = { - k: str(v)[:20] for k, v in std_params.__dict__.items() if v + k: str(v)[:30] for k, v in std_params.__dict__.items() if v } standard_params_json = json.dumps(standard_params) return standard_params_json + def _handle_data_repr(obbject: OBBject) -> str: + """Handle data representation for obbjects""" + data_repr = "" + if hasattr(obbject, "results") and obbject.results: + data_schema = ( + obbject.results[0].model_json_schema() + if obbject.results and isinstance(obbject.results, list) + else "" + ) + if data_schema and "title" in data_schema: + data_repr = f"{data_schema['title']}" + if data_schema and "description" in data_schema: + data_repr += f" - {data_schema['description'].split('.')[0]}" + + return data_repr + obbjects = {} - for obbject in self.obbject_instances: - obbjects[obbject.id] = { + for i, obbject in enumerate(list(reversed(self.obbjects))): + obbjects[i] = { "route": obbject._route, # pylint: disable=protected-access "provider": obbject.provider, + "standard params": _handle_standard_params(obbject), + "data": _handle_data_repr(obbject), } - data_schema = ( - obbject.results[0].model_json_schema() - if obbject.results and isinstance(obbject.results, list) - else "" - ) - data_repr = ( - f"{data_schema['title']} - {data_schema['description'].split('.')[0]}" - ) - obbjects[obbject.id]["standard params"] = _handle_standard_params(obbject) - obbjects[obbject.id]["data"] = data_repr - return obbjects diff --git a/openbb_terminal/base_platform_controller.py b/openbb_terminal/base_platform_controller.py index bc8b1b0ab61..5fe15763ff5 100644 --- a/openbb_terminal/base_platform_controller.py +++ b/openbb_terminal/base_platform_controller.py @@ -79,17 +79,15 @@ class PlatformController(BaseController): for _, trl in self.translators.items(): for action in trl._parser._actions: # pylint: disable=protected-access if action.dest == "data": - # this is purposely making the choices pointing to the registry ids - action.choices = Registry.ids - # also change the action type to str - action.type = str + action.choices = range(len(Registry.obbjects)) + action.type = int def _intersect_data_processing_commands(self, ns_parser): """Intersect data processing commands and change the obbject id into an actual obbject.""" if ( hasattr(ns_parser, "data") and ns_parser.data - and ns_parser.data in Registry.ids + and ns_parser.data in range(len(Registry.obbjects)) ): obbject = Registry.get(ns_parser.data) setattr(ns_parser, "data", obbject.results) diff --git a/openbb_terminal/terminal_controller.py b/openbb_terminal/terminal_controller.py index 9823ca3a10a..c7651a525f4 100644 --- a/openbb_terminal/terminal_controller.py +++ b/openbb_terminal/terminal_controller.py @@ -463,7 +463,7 @@ class TerminalController(BaseController): if results: df = pd.DataFrame.from_dict(results, orient="index") print_rich_table( - df, show_index=True, index_name="uuid", title="OBBject Results" + df, show_index=True, index_name="stack index", title="OBBject Results" ) else: console.print("[info]No results found.[/info]") |