summaryrefslogtreecommitdiffstats
diff options
context:
space:
mode:
authorhjoaquim <h.joaquim@campus.fct.unl.pt>2024-04-23 11:15:22 +0100
committerhjoaquim <h.joaquim@campus.fct.unl.pt>2024-04-23 11:15:22 +0100
commitf3b82d6e1d81ea2871f77231a541f1ba448a05c3 (patch)
treedd633a005e3257fda80f5eb59c99873950b701ba
parent0c2a92e86efa507dfdf393dec2d00c528f7f09b1 (diff)
improved obbject registry using a stack
-rw-r--r--argparse_translator/obbject_registry.py85
-rw-r--r--openbb_terminal/base_platform_controller.py8
-rw-r--r--openbb_terminal/terminal_controller.py2
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]")