jerry/OpenBB
1
0
Fork 0
mirror of https://github.com/OpenBB-finance/OpenBB.git synced 2026-10-27 04:31:13 +00:00
Code Issues Packages Projects Releases Wiki Activity

[Feature] Remove Account Module and HubService (#7236)

Browse Source 
* remove account module

* unused import

* fix test module
This commit is contained in:
Danglewood
2025-10-11 15:53:06 -07:00
committed by GitHub
parent 9f0d592839
commit 467a2f3b1a
27 changed files with 70 additions and 1637 deletions
Download Patch File Download Diff File Expand all files Collapse all files

30
cli/README.md vendored
View File

@@ -36,32 +36,6 @@ Which should result in the following output:
![image](https://github.com/OpenBB-finance/OpenBB/assets/48914296/f606bb6e-fa00-4fc8-bad2-8269bb4fc38e)
## API keys
## Documentation
To fully leverage the OpenBB Platform you need to get some API keys to connect with data providers. Here are the 3 options on where to set them:
1. OpenBB Hub
2. Local file
### 1. OpenBB Hub
Set your keys at [OpenBB Hub](https://my.openbb.co/app/platform/credentials) and get your personal access token from <https://my.openbb.co/app/platform/pat> to connect with your account.
> Once you log in, on the Platform CLI (through the `/account` menu, all your credentials will be in sync with the OpenBB Hub.)
### 2. Local file
You can specify the keys directly in the `~/.openbb_platform/user_settings.json` file.
Populate this file with the following template and replace the values with your keys:
```json
{
"credentials": {
"fmp_api_key": "REPLACE_ME",
"polygon_api_key": "REPLACE_ME",
"benzinga_api_key": "REPLACE_ME",
"fred_api_key": "REPLACE_ME"
}
}
```
View the user documentation for this package [here](https://docs.openbb.co/cli)

60
cli/integration/test_integration_hub_service.py
View File

@@ -1,60 +0,0 @@
"""Integration tests for the hub_service module."""
from unittest.mock import create_autospec, patch
import pytest
import requests
from openbb_cli.controllers.hub_service import upload_routine
from openbb_core.app.model.hub.hub_session import HubSession
# pylint: disable=unused-argument, redefined-outer-name, unused-variable
@pytest.fixture
def auth_header():
"""Return a fake auth header."""
return "Bearer fake_token"
@pytest.fixture
def hub_session_mock():
"""Return a mock HubSession."""
mock = create_autospec(HubSession, instance=True)
mock.username = "TestUser"
return mock
# Fixture for routine data
@pytest.fixture
def routine_data():
"""Return a dictionary with routine data."""
return {
"name": "Test Routine",
"description": "A test routine",
"routine": "print('Hello World')",
"override": False,
"tags": "test",
"public": True,
}
@pytest.mark.integration
def test_upload_routine_timeout(auth_header, routine_data):
"""Test upload_routine with a timeout exception."""
with patch(
"requests.post", side_effect=requests.exceptions.Timeout
) as mocked_post: # noqa: F841
response = upload_routine(auth_header, **routine_data)
assert response is None
@pytest.mark.integration
def test_upload_routine_connection_error(auth_header, routine_data):
"""Test upload_routine with a connection error."""
with patch(
"requests.post", side_effect=requests.exceptions.ConnectionError
) as mocked_post: # noqa: F841
response = upload_routine(auth_header, **routine_data)
assert response is None

184
cli/openbb_cli/controllers/base_controller.py
View File

@@ -14,16 +14,13 @@ import pandas as pd
from openbb_cli.config.completer import NestedCompleter
from openbb_cli.config.constants import SCRIPT_TAGS
from openbb_cli.controllers.choices import build_controller_choice_map
from openbb_cli.controllers.hub_service import upload_routine
from openbb_cli.controllers.utils import (
check_file_type_saved,
check_positive,
get_flair_and_username,
handle_obbject_display,
parse_unknown_args_to_dict,
print_guest_block_msg,
print_rich_table,
remove_file,
system_clear,
validate_register_key,
)
@@ -40,7 +37,6 @@ session = Session()
# TODO: We should try to avoid these global variables
RECORD_SESSION = False
RECORD_SESSION_LOCAL_ONLY = False
SESSION_RECORDED = list()
SESSION_RECORDED_NAME = ""
SESSION_RECORDED_DESCRIPTION = ""
@@ -65,7 +61,6 @@ class BaseController(metaclass=ABCMeta):
"r",
"reset",
"stop",
"whoami",
"results",
]
@@ -271,11 +266,6 @@ class BaseController(metaclass=ABCMeta):
for _ in range(self.PATH.count("/")):
self.queue.insert(0, "quit")
if not session.is_local():
remove_file(
Path(session.user.preferences.export_directory, "routines", "hub")
)
def call_reset(self, _) -> None:
"""Process reset command.
@@ -398,17 +388,6 @@ class BaseController(metaclass=ABCMeta):
)
return
if session.is_local():
session.console.print(
"[red]Recording session to the OpenBB Hub is not supported in guest mode.[/red]"
)
session.console.print(
"\n[yellow]Visit the OpenBB Hub to register: http://my.openbb.co[/yellow]"
)
session.console.print(
"\n[yellow]Your routine will be saved locally.[/yellow]\n"
)
# Check if title has a valid format
title = " ".join(ns_parser.name) if ns_parser.name else ""
pattern = re.compile(r"^[a-zA-Z0-9\s]+$")
@@ -419,13 +398,11 @@ class BaseController(metaclass=ABCMeta):
return
global RECORD_SESSION # noqa: PLW0603
global RECORD_SESSION_LOCAL_ONLY # noqa: PLW0603
global SESSION_RECORDED_NAME # noqa: PLW0603
global SESSION_RECORDED_DESCRIPTION # noqa: PLW0603
global SESSION_RECORDED_TAGS # noqa: PLW0603
global SESSION_RECORDED_PUBLIC # noqa: PLW0603
RECORD_SESSION_LOCAL_ONLY = session.is_local()
RECORD_SESSION = True
SESSION_RECORDED_NAME = title
SESSION_RECORDED_DESCRIPTION = (
@@ -471,135 +448,64 @@ class BaseController(metaclass=ABCMeta):
)
else:
current_user = session.user
title_for_local_storage = (
SESSION_RECORDED_NAME.replace(" ", "_") + ".openbb"
)
# 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"
routine_file = os.path.join(
f"{current_user.preferences.export_directory}/routines",
title_for_local_storage,
)
# 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,
)
# 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("")
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"
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"
)
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)
# Writing to file
Path(os.path.dirname(routine_file)).mkdir(parents=True, exist_ok=True)
session.console.print(
f"[green]Your routine has been recorded and saved here: {routine_file}[/green]\n"
)
with open(routine_file, "w") as file1:
lines = ["# OpenBB Platform CLI - Routine", "\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 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]")
session.console.print(
f"[green]Your routine has been recorded and saved here: {routine_file}[/green]\n"
)
# Clear session to be recorded again
RECORD_SESSION = False
SESSION_RECORDED = list()
def call_whoami(self, other_args: list[str]) -> None:
"""Process whoami command."""
parser = argparse.ArgumentParser(
add_help=False,
formatter_class=argparse.ArgumentDefaultsHelpFormatter,
prog="whoami",
description="Show current user",
)
ns_parser, _ = self.parse_simple_args(parser, other_args)
if ns_parser:
current_user = session.user
local_user = session.is_local()
if not local_user:
hub_session = current_user.profile.hub_session
session.console.print(
f"[info]email:[/info] {hub_session.email if hub_session else 'N/A'}"
)
session.console.print(
f"[info]uuid:[/info] {hub_session.user_uuid if hub_session else 'N/A'}"
)
else:
print_guest_block_msg()
def call_results(self, other_args: list[str]):
"""Process results command."""
parser = argparse.ArgumentParser(

20
cli/openbb_cli/controllers/cli_controller.py
View File

@@ -55,7 +55,7 @@ PLATFORM_ROUTERS = {
for d in dir(obb)
if "_" not in d
}
NON_DATA_ROUTERS = ["coverage", "account", "reference", "system", "user"]
NON_DATA_ROUTERS = ["coverage", "reference", "system", "user"]
DATA_PROCESSING_ROUTERS = ["technical", "quantitative", "econometrics"]
env_file = str(ENV_FILE_SETTINGS)
session = Session()
@@ -216,23 +216,7 @@ class CLIController(BaseController):
def print_help(self):
"""Print help."""
mt = MenuText("")
mt.add_info("Configure the platform and manage your account")
for router, value in PLATFORM_ROUTERS.items():
if router not in NON_DATA_ROUTERS or router in ["reference", "coverage"]:
continue
if value == "menu":
menu_description = (
obb.reference["routers"].get(f"{self.PATH}{router}", {}).get("description") # type: ignore
) or ""
mt.add_menu(
name=router,
description=menu_description.split(".")[0].lower(),
)
else:
mt.add_cmd(router)
mt.add_info("\nConfigure your CLI")
mt.add_info("\nConfigure CLI")
mt.add_menu(
"settings",
description="enable and disable feature flags, preferences and settings",

105
cli/openbb_cli/controllers/hub_service.py
View File

@@ -1,105 +0,0 @@
"""Routines handler module."""
import requests
from openbb_cli.config.constants import (
CONNECTION_ERROR_MSG,
CONNECTION_TIMEOUT_MSG,
TIMEOUT,
)
from openbb_cli.session import Session
# created dictionaries for personal and default routines with the structure
# {"file_name" :["script","personal/default"]}
# and stored dictionaries in list
# created new directory structure to account for personal and default routines
session = Session()
# pylint: disable=too-many-arguments, too-many-positional-arguments
def upload_routine(
auth_header: str,
name: str = "",
description: str = "",
routine: str = "",
override: bool = False,
tags: str = "",
public: bool = False,
timeout: int = TIMEOUT,
) -> requests.Response | None:
"""Send a routine to the server.
Parameters
----------
auth_header : str
The authorization header, e.g. "Bearer <token>".
name : str
The name of the routine.
routine : str
The routine.
override : bool
Whether to override the routine if it already exists.
tags : str
The tags of the routine.
public : bool
Whether to make the routine public or not.
timeout : int
The timeout, by default TIMEOUT
Returns
-------
Optional[requests.Response]
The response from the post request.
"""
data = {
"name": name,
"description": description,
"script": routine,
"override": override,
"tags": tags,
"version": session.settings.VERSION,
"public": public,
}
_console = session.console
try:
response = requests.post(
headers={"Authorization": auth_header},
url=session.settings.BASE_URL + "/terminal/script",
json=data,
timeout=timeout,
)
if response.status_code == 200:
username = getattr(session.user.profile.hub_session, "username", None)
if not username:
_console.print("[red]No username found.[/red]")
_console.print("[red]Failed to upload your routine.[/red]")
return None
_console.print("[green]Successfully uploaded your routine.[/]")
hub_url = session.settings.HUB_URL
if public:
_console.print(
f"\n[yellow]Share or edit it at {hub_url}/u/{username}/routine/{name.replace(' ', '-')}[/]"
)
else:
_console.print(f"\n[yellow]Go to {hub_url} to edit this script,[/]")
_console.print(
f"[yellow]or even make it public so you can access it at "
f"{hub_url}/u/{username}/routine/{name.replace(' ', '-')}[/]"
)
elif response.status_code != 409: # 409: routine already exists
_console.print(
"[red]" + response.json().get("detail", "Unknown error.") + "[/red]"
)
return response
except requests.exceptions.ConnectionError:
_console.print(f"\n{CONNECTION_ERROR_MSG}")
return None
except requests.exceptions.Timeout:
_console.print(f"\n{CONNECTION_TIMEOUT_MSG}")
return None
except Exception:
_console.print("[red]Failed to upload your routine.[/red]")
return None

44
cli/openbb_cli/controllers/utils.py
View File

@@ -65,29 +65,17 @@ def print_goodbye():
text = """
[param]Thank you for using the OpenBB Platform CLI and being part of this journey.[/param]
We hope you'll find the new OpenBB Platform CLI a valuable tool.
To stay tuned, sign up for our newsletter: [cmds]https://openbb.co/newsletter.[/]
Please feel free to check out our other products:
[bold]OpenBB Workspace[/]: [cmds]https://openbb.co[/cmds]
[bold]OpenBB Platform:[/] [cmds]https://docs.openbb.co/platform[/cmds]
"""
[bold]ODP Desktop Application:[/] [cmds]https://docs.openbb.co/odp/[/cmds]
[bold]ODP Python Package:[/] [cmds]https://docs.openbb.co/platform[/cmds]
"""
session.console.print(text)
def print_guest_block_msg():
"""Block guest users from using the cli."""
if session.is_local():
session.console.print(
"[info]You are currently logged as a guest.[/info]\n"
"[info]Login to use this feature.[/info]\n\n"
"[info]If you don't have an account, you can create one here: [/info]"
f"[cmds]{session.settings.HUB_URL + '/register'}\n[/cmds]"
)
def bootup():
"""Bootup the cli."""
if sys.platform == "win32":
@@ -124,15 +112,6 @@ def reset(queue: list[str] | None = None):
dev = session.settings.DEV_BACKEND
try:
# remove the hub routines
if not session.is_local():
remove_file(
Path(session.user.preferences.export_directory, "routines", "hub")
)
# if not get_current_user().profile.remember:
# Local.remove(HIST_FILE_PROMPT)
# we clear all openbb_cli modules from sys.modules
for module in list(sys.modules.keys()):
parts = module.split(".")
@@ -140,16 +119,11 @@ def reset(queue: list[str] | None = None):
del sys.modules[module]
queue_list = ["/".join(queue) if len(queue) > 0 else ""] # type: ignore
# pylint: disable=import-outside-toplevel
# we run the cli again
if session.is_local():
from openbb_cli.controllers.cli_controller import main
from openbb_cli.controllers.cli_controller import main
main(debug, dev, queue_list, module="") # type: ignore
else:
from openbb_cli.controllers.cli_controller import launch
launch(queue=queue_list)
main(debug, dev, queue_list, module="") # type: ignore
except Exception as e:
session.console.print(f"Unfortunately, resetting wasn't possible: {e}\n")
@@ -550,11 +524,7 @@ def get_flair_and_username() -> str:
if dtime := get_dtime():
dtime = f"{dtime} "
username = getattr(session.user.profile.hub_session, "username", "")
if username:
username = f"[{username}] "
return f"{dtime}{username}{flair}"
return f"{dtime}{flair}"
def is_timezone_valid(user_tz: str) -> bool:

6
cli/openbb_cli/session.py
View File

@@ -25,7 +25,7 @@ def _get_backend():
except ValueError:
# backend might not be created yet
charting_settings = ChartingSettings(
system_settings=obb.system,
system_settings=obb.system, # type: ignore
user_settings=obb.user, # type: ignore
)
create_backend(charting_settings)
@@ -97,10 +97,6 @@ class Session(metaclass=SingletonMeta):
return prompt_session
def is_local(self) -> bool:
"""Check if user is local."""
return not bool(self.user.profile.hub_session)
def max_obbjects_exceeded(self) -> bool:
"""Check if max obbjects exceeded."""
return (

10
cli/tests/test_controllers_utils.py
View File

@@ -12,7 +12,6 @@ from openbb_cli.controllers.utils import (
get_user_agent,
parse_and_split_input,
print_goodbye,
print_guest_block_msg,
remove_file,
welcome_message,
)
@@ -25,9 +24,7 @@ def mock_session():
"""Mock the session and its dependencies."""
with patch("openbb_cli.controllers.utils.session") as mock_session:
mock_session.console.print = MagicMock()
mock_session.is_local = MagicMock(return_value=True)
mock_session.settings.VERSION = "1.0"
mock_session.user.profile.hub_session.username = "testuser"
mock_session.settings.FLAIR = "rocket"
yield mock_session
@@ -87,12 +84,6 @@ def test_parse_and_split_input_special_cases(input_command, expected_output):
assert result == expected_output
def test_print_guest_block_msg(mock_session):
"""Test printing the guest block message."""
print_guest_block_msg()
mock_session.console.print.assert_called()
def test_welcome_message(mock_session):
"""Test printing the welcome message."""
welcome_message()
@@ -104,7 +95,6 @@ def test_welcome_message(mock_session):
def test_get_flair_and_username(mock_session):
"""Test getting the flair and username."""
result = get_flair_and_username()
assert "testuser" in result
assert "rocket" in result

25
openbb_platform/core/openbb_core/app/model/hub/hub_session.py
View File

@@ -1,25 +0,0 @@
"""Model for HubSession."""
from pydantic import BaseModel, SecretStr, field_serializer
class HubSession(BaseModel):
"""Model for HubSession."""
username: str | None = None
email: str
primary_usage: str
user_uuid: str
token_type: str
access_token: SecretStr
def __repr__(self) -> str:
"""Return string representation."""
return f"{self.__class__.__name__}\n\n" + "\n".join(
f"{k}: {v}" for k, v in self.model_dump().items()
)
@field_serializer("access_token", when_used="json-unless-none")
def _dump_secret(self, v):
"""Dump secret."""
return v.get_secret_value()

22
openbb_platform/core/openbb_core/app/model/hub/hub_user_settings.py
View File

@@ -1,22 +0,0 @@
"""Hub user settings model."""
from typing import Any
from pydantic import BaseModel, ConfigDict, Field, field_validator
class HubUserSettings(BaseModel):
"""Hub user settings model."""
features_settings: dict[str, Any] = Field(default_factory=dict)
features_keys: dict[str, str | None] = Field(default_factory=dict)
# features_sources: Dict[str, Any]
# features_terminal_style: Dict[str, Union[str, Dict[str, str]]]
model_config = ConfigDict(validate_assignment=True)
@field_validator("features_keys", mode="before", check_fields=False)
@classmethod
def to_lower(cls, d: dict) -> dict:
"""Convert dict keys to lowercase."""
return {k.lower(): v for k, v in d.items()}

17
openbb_platform/core/openbb_core/app/model/profile.py
View File

@@ -1,17 +0,0 @@
"""Profile model."""
from openbb_core.app.model.hub.hub_session import HubSession
from pydantic import BaseModel, ConfigDict, Field
class Profile(BaseModel):
"""Profile."""
hub_session: HubSession | None = Field(default=None)
model_config = ConfigDict(validate_assignment=True)
def __repr__(self) -> str:
"""Return string representation."""
return f"{self.__class__.__name__}\n\n" + "\n".join(
f"{k}: {v}" for k, v in self.model_dump().items()
)

2
openbb_platform/core/openbb_core/app/model/user_settings.py
View File

@@ -9,14 +9,12 @@ from openbb_core.app.model.abstract.tagged import Tagged
from openbb_core.app.model.credentials import Credentials
from openbb_core.app.model.defaults import Defaults
from openbb_core.app.model.preferences import Preferences
from openbb_core.app.model.profile import Profile
from pydantic import Field
class UserSettings(Tagged):
"""User settings."""
profile: Profile = Field(default_factory=Profile)
credentials: Credentials = Field(default_factory=Credentials)
preferences: Preferences = Field(default_factory=Preferences)
defaults: Defaults = Field(default_factory=Defaults)

287
openbb_platform/core/openbb_core/app/service/hub_service.py
View File

@@ -1,287 +0,0 @@
"""Hub manager class."""
from warnings import warn
from fastapi import HTTPException
from jwt import ExpiredSignatureError, PyJWTError, decode, get_unverified_header
from openbb_core.app.model.abstract.error import OpenBBError
from openbb_core.app.model.credentials import Credentials
from openbb_core.app.model.defaults import Defaults
from openbb_core.app.model.hub.hub_session import HubSession
from openbb_core.app.model.hub.hub_user_settings import HubUserSettings
from openbb_core.app.model.profile import Profile
from openbb_core.app.model.user_settings import UserSettings
from openbb_core.env import Env
class HubService:
"""Hub service class."""
TIMEOUT = 10
# Mapping of V3 keys to V4 keys for backward compatibility
V3TOV4 = {
"api_key_alphavantage": "alpha_vantage_api_key",
"api_biztoc_token": "biztoc_api_key",
"api_fred_key": "fred_api_key",
"api_key_financialmodelingprep": "fmp_api_key",
"api_intrinio_key": "intrinio_api_key",
"api_polygon_key": "polygon_api_key",
"api_key_quandl": "nasdaq_api_key",
"api_tradier_token": "tradier_api_key",
}
V4TOV3 = {v: k for k, v in V3TOV4.items()}
def __init__(
self,
session: HubSession | None = None,
base_url: str | None = None,
):
"""Initialize Hub service."""
# pylint: disable=import-outside-toplevel
from openbb_core.provider.utils.helpers import get_requests_session
self._base_url = base_url or Env().HUB_BACKEND
self._session = session
self._hub_user_settings: HubUserSettings | None = None
self._request_session = get_requests_session()
@property
def base_url(self) -> str:
"""Get base url."""
return self._base_url
@property
def session(self) -> HubSession | None:
"""Get session."""
return self._session
def connect(
self,
email: str | None = None,
password: str | None = None,
pat: str | None = None,
) -> HubSession:
"""Connect to Hub."""
if email and password:
self._session = self._get_session_from_email_password(email, password)
return self._session
if pat:
self._session = self._get_session_from_platform_token(pat)
return self._session
raise OpenBBError("Please provide 'email' and 'password' or 'pat'")
def disconnect(self) -> bool:
"""Disconnect from Hub."""
if self._session:
result = self._post_logout(self._session)
self._session = None
return result
raise OpenBBError(
"No session found. Login or provide a 'HubSession' on initialization."
)
def push(self, user_settings: UserSettings) -> bool:
"""Push user settings to Hub."""
if self._session:
if user_settings.credentials:
hub_user_settings = self.platform2hub(
user_settings.credentials, user_settings.defaults
)
return self._put_user_settings(self._session, hub_user_settings)
return False
raise OpenBBError(
"No session found. Login or provide a 'HubSession' on initialization."
)
def pull(self) -> UserSettings:
"""Pull user settings from Hub."""
if self._session:
self._hub_user_settings = self._get_user_settings(self._session)
profile = Profile(hub_session=self._session)
credentials, defaults = self.hub2platform(self._hub_user_settings)
return UserSettings(
profile=profile, credentials=credentials, defaults=defaults
)
raise OpenBBError(
"No session found. Login or provide a 'HubSession' on initialization."
)
def _get_session_from_email_password(self, email: str, password: str) -> HubSession:
"""Get session from email and password."""
if not email:
raise OpenBBError("Email not found.")
if not password:
raise OpenBBError("Password not found.")
response = self._request_session.post(
url=self._base_url + "/login",
json={
"email": email,
"password": password,
"remember": True,
},
timeout=self.TIMEOUT,
)
if response.status_code == 200:
session = response.json()
hub_session = HubSession(
access_token=session.get("access_token"),
token_type=session.get("token_type"),
user_uuid=session.get("uuid"),
email=session.get("email"),
username=session.get("username"),
primary_usage=session.get("primary_usage"),
)
return hub_session
status_code = response.status_code
detail = response.json().get("detail", None)
raise HTTPException(status_code, detail)
def _get_session_from_platform_token(self, token: str) -> HubSession:
"""Get session from Platform personal access token."""
if not token:
raise OpenBBError("Platform personal access token not found.")
self._check_token_expiration(token)
response = self._request_session.post(
url=self._base_url + "/sdk/login",
json={
"token": token,
},
timeout=self.TIMEOUT,
)
if response.status_code == 200:
session = response.json()
hub_session = HubSession(
access_token=session.get("access_token"),
token_type=session.get("token_type"),
user_uuid=session.get("uuid"),
username=session.get("username"),
email=session.get("email"),
primary_usage=session.get("primary_usage"),
)
return hub_session
status_code = response.status_code
detail = response.json().get("detail", None)
raise HTTPException(status_code, detail)
def _post_logout(self, session: HubSession) -> bool:
"""Post logout."""
access_token = session.access_token.get_secret_value()
token_type = session.token_type
authorization = f"{token_type.title()} {access_token}"
response = self._request_session.get(
url=self._base_url + "/logout",
headers={"Authorization": authorization},
json={"token": access_token},
timeout=self.TIMEOUT,
)
if response.status_code == 200:
result = response.json()
return result.get("success", False)
status_code = response.status_code
result = response.json()
detail = result.get("detail", None)
raise HTTPException(status_code, detail)
def _get_user_settings(self, session: HubSession) -> HubUserSettings:
"""Get user settings."""
access_token = session.access_token.get_secret_value()
token_type = session.token_type
authorization = f"{token_type.title()} {access_token}"
response = self._request_session.get(
url=self._base_url + "/terminal/user",
headers={"Authorization": authorization},
timeout=self.TIMEOUT,
)
if response.status_code == 200:
user_settings = response.json()
filtered = {k: v for k, v in user_settings.items() if v is not None}
return HubUserSettings.model_validate(filtered)
status_code = response.status_code
detail = response.json().get("detail", None)
raise HTTPException(status_code, detail)
def _put_user_settings(
self, session: HubSession, settings: HubUserSettings
) -> bool:
"""Put user settings."""
access_token = session.access_token.get_secret_value()
token_type = session.token_type
authorization = f"{token_type.title()} {access_token}"
response = self._request_session.put(
url=self._base_url + "/user",
headers={"Authorization": authorization},
json=settings.model_dump(exclude_defaults=True),
timeout=self.TIMEOUT,
)
if response.status_code == 200:
return True
status_code = response.status_code
detail = response.json().get("detail", None)
raise HTTPException(status_code, detail)
def hub2platform(self, settings: HubUserSettings) -> tuple[Credentials, Defaults]:
"""Convert Hub user settings to Platform models."""
deprecated = {
k: v for k, v in self.V3TOV4.items() if k in settings.features_keys
}
if deprecated:
msg = ""
for k, v in deprecated.items():
msg += f"\n'{k.upper()}' -> '{v.upper()}', "
msg = msg.strip(", ")
warn(
message=f"\nDeprecated v3 credentials found.\n{msg}"
"\n\nYou can update them at https://my.openbb.co/app/platform/credentials.",
)
# We give priority to v4 keys over v3 keys if both are present
hub_credentials = {
self.V3TOV4.get(k, k): settings.features_keys.get(self.V3TOV4.get(k, k), v)
for k, v in settings.features_keys.items()
}
defaults = settings.features_settings.get("defaults", {})
return Credentials(**hub_credentials), Defaults(**defaults)
def platform2hub(
self, credentials: Credentials, defaults: Defaults
) -> HubUserSettings:
"""Convert Platform models to Hub user settings."""
# Dump mode json ensures SecretStr values are serialized as strings
credentials = credentials.model_dump(
mode="json", exclude_none=True, exclude_defaults=True
)
settings = self._hub_user_settings or HubUserSettings()
for v4_k, v in sorted(credentials.items()):
v3_k = self.V4TOV3.get(v4_k, None)
# If v3 key was in the hub already, we keep it
k = v3_k if v3_k in settings.features_keys else v4_k
settings.features_keys[k] = v
defaults_ = defaults.model_dump(
mode="json", exclude_none=True, exclude_defaults=True
)
settings.features_settings.update({"defaults": defaults_})
return settings
@staticmethod
def _check_token_expiration(token: str) -> None:
"""Check token expiration, raises exception if expired."""
try:
header_data = get_unverified_header(token)
decode(
token,
key="secret",
algorithms=[header_data["alg"]],
options={"verify_signature": False, "verify_exp": True},
)
except ExpiredSignatureError as e:
raise OpenBBError("Platform personal access token expired.") from e
except PyJWTError as e:
raise OpenBBError("Failed to decode Platform token.") from e

243
openbb_platform/core/openbb_core/app/static/account.py
View File

@@ -1,243 +0,0 @@
"""Account."""
# pylint: disable=W0212:protected-access
import json
import logging
from functools import wraps
from pathlib import Path
from sys import exc_info
from typing import TYPE_CHECKING
from openbb_core.app.logs.logging_service import LoggingService
from openbb_core.app.model.abstract.error import OpenBBError
from openbb_core.app.model.hub.hub_session import HubSession
from openbb_core.app.model.user_settings import UserSettings
from openbb_core.app.service.hub_service import HubService
from openbb_core.app.service.user_service import UserService
if TYPE_CHECKING:
from openbb_core.app.static.app_factory import BaseApp
logger = logging.getLogger(__name__)
logger.addHandler(logging.StreamHandler())
logger.setLevel(logging.WARN)
dep_warning = (
"\nDeprecation Warning: The Account module is deprecated and will be removed in a future version."
+ " Please migrate to using the `user_settings.json` file. For more information, visit:"
+ " https://docs.openbb.co/platform/settings/user_settings/api_keys"
)
class Account: # noqa: D205, D400
"""/account
login
logout
save
refresh
"""
SESSION_FILE = ".hub_session.json"
def __init__(self, base_app: "BaseApp"):
"""Initialize account service."""
self._base_app = base_app
self._openbb_directory = (
base_app._command_runner.system_settings.openbb_directory
)
self._hub_service: HubService | None = None
def __repr__(self) -> str:
"""Human readable representation of the object."""
return self.__doc__ or ""
def _log_account_command(func): # pylint: disable=E0213
"""Log account command."""
@wraps(func) # type: ignore[arg-type]
def wrapped(self, *args, **kwargs):
try:
# pylint: disable=E1102
result = func(self, *args, **kwargs) # type: ignore[operator]
except Exception as e:
raise OpenBBError(e) from e
finally:
user_settings = self._base_app._command_runner.user_settings
system_settings = self._base_app._command_runner.system_settings
if system_settings.logging_suppress is False:
ls = LoggingService(
user_settings=user_settings, system_settings=system_settings
)
ls.log(
user_settings=user_settings,
system_settings=system_settings,
# pylint: disable=E1101
route=f"/account/{func.__name__}", # type: ignore[attr-defined]
func=func, # type: ignore[arg-type]
kwargs={}, # don't want any credentials being logged by accident
exec_info=exc_info(),
)
return result
return wrapped
def _create_hub_service(
self,
email: str | None = None,
password: str | None = None,
pat: str | None = None,
) -> HubService:
"""Create hub service to handle connection."""
if email is None and password is None and pat is None:
session_file = Path(self._openbb_directory, self.SESSION_FILE)
if not session_file.exists():
raise OpenBBError("Session not found.")
with open(session_file) as f:
session_dict = json.load(f)
hub_session = HubSession(**session_dict)
hs = HubService(hub_session)
else:
hs = HubService()
hs.connect(email, password, pat)
return hs
# pylint: disable=R0917
@_log_account_command # type: ignore
def login(
self,
email: str | None = None,
password: str | None = None,
pat: str | None = None,
remember_me: bool = False,
return_settings: bool = False,
) -> UserSettings | None:
"""Login to hub.
Parameters
----------
email : Optional[str], optional
Email address, by default None
password : Optional[str], optional
Password, by default None
pat : Optional[str], optional
Personal access token, by default None
remember_me : bool, optional
Remember me, by default False
return_settings : bool, optional
Return user settings, by default False
Returns
-------
Optional[UserSettings]
User settings: profile, credentials, preferences
"""
logger.warning(dep_warning)
self._hub_service = self._create_hub_service(email, password, pat)
incoming = self._hub_service.pull()
self._base_app.user.profile = incoming.profile
self._base_app.user.credentials.update(incoming.credentials)
self._base_app.user.defaults.update(incoming.defaults)
if remember_me:
Path(self._openbb_directory).mkdir(parents=False, exist_ok=True)
session_file = Path(self._openbb_directory, self.SESSION_FILE)
with open(session_file, "w") as f:
if not self._hub_service.session:
raise OpenBBError("Not connected to hub.")
json.dump(
self._hub_service.session.model_dump(mode="json"), f, indent=4
)
self._base_app._command_runner.user_settings.profile.hub_session = getattr(
self._base_app._account._hub_service, "_session", None
)
if return_settings:
return self._base_app._command_runner.user_settings
return None
@_log_account_command # type: ignore
def save(self, return_settings: bool = False) -> UserSettings | None:
"""Save user settings.
Parameters
----------
return_settings : bool, optional
Return user settings, by default False
Returns
-------
Optional[UserSettings]
User settings: profile, credentials, preferences
"""
logger.warning(dep_warning)
if not self._hub_service:
UserService.write_to_file(self._base_app._command_runner.user_settings)
else:
self._hub_service.push(self._base_app._command_runner.user_settings)
if return_settings:
return self._base_app._command_runner.user_settings
return None
@_log_account_command # type: ignore
def refresh(self, return_settings: bool = False) -> UserSettings | None:
"""Refresh user settings.
Parameters
----------
return_settings : bool, optional
Return user settings, by default False
Returns
-------
Optional[UserSettings]
User settings: profile, credentials, preferences
"""
logger.warning(dep_warning)
if not self._hub_service:
self._base_app._command_runner.user_settings = UserService.read_from_file()
else:
incoming = self._hub_service.pull()
self._base_app.user.profile = incoming.profile
self._base_app.user.credentials.update(incoming.credentials)
self._base_app.user.defaults.update(incoming.defaults)
if return_settings:
return self._base_app._command_runner.user_settings
return None
@_log_account_command # type: ignore
def logout(self, return_settings: bool = False) -> UserSettings | None:
"""Logout from hub.
Parameters
----------
return_settings : bool, optional
Return user settings, by default False
Returns
-------
Optional[UserSettings]
User settings: profile, credentials, preferences
"""
logger.warning(dep_warning)
if not self._hub_service:
raise OpenBBError("Not connected to hub.")
self._hub_service.disconnect()
session_file = Path(self._openbb_directory, self.SESSION_FILE)
if session_file.exists():
session_file.unlink()
self._base_app._command_runner.user_settings = UserService.read_from_file()
if return_settings:
return self._base_app._command_runner.user_settings
return None

12
openbb_platform/core/openbb_core/app/static/app_factory.py
View File

@@ -5,7 +5,6 @@ from typing import TypeVar
from openbb_core.app.command_runner import CommandRunner
from openbb_core.app.model.system_settings import SystemSettings
from openbb_core.app.model.user_settings import UserSettings
from openbb_core.app.static.account import Account
from openbb_core.app.static.container import Container
from openbb_core.app.static.coverage import Coverage
from openbb_core.app.static.reference_loader import ReferenceLoader
@@ -15,7 +14,6 @@ E = TypeVar("E", bound=type[Container])
BASE_DOC = f"""OpenBB Platform v{VERSION}
Utilities:
/account
/user
/system
/coverage
@@ -29,15 +27,9 @@ class BaseApp:
"""Initialize the app."""
command_runner.init_logging_service()
self._command_runner = command_runner
self._account = Account(self)
self._coverage = Coverage(self)
self._reference = ReferenceLoader().reference
@property
def account(self) -> Account:
"""Account menu."""
return self._account
@property
def user(self) -> UserSettings:
"""User settings."""
@@ -59,7 +51,7 @@ class BaseApp:
return self._reference
def create_app(extensions: E | None = None) -> type[BaseApp]:
def create_app(extensions: E | None = None) -> type[BaseApp]: # type: ignore
"""Create the app."""
class App(BaseApp, extensions or object): # type: ignore[misc]
@@ -68,4 +60,4 @@ def create_app(extensions: E | None = None) -> type[BaseApp]:
ext_doc = extensions.__doc__ if extensions else ""
return BASE_DOC + (ext_doc or "")
return App(command_runner=CommandRunner())
return App(command_runner=CommandRunner()) # type: ignore[call-arg]

1
openbb_platform/core/tests/app/model/hub/__init__.py
View File

@@ -1 +0,0 @@
"""OpenBB Platform Core tests."""

37
openbb_platform/core/tests/app/model/hub/test_hub_session.py
View File

@@ -1,37 +0,0 @@
"""Test the HubSession class."""
from openbb_core.app.model.hub.hub_session import HubSession
from pydantic import SecretStr
# ruff: noqa: S105 S106
def test_hub_session():
"""Test the HubSession class."""
session = HubSession(
access_token=SecretStr("mock_access_token"),
token_type="mock_token_type",
email="mock_email",
user_uuid="mock_user_uuid",
username="mock_username",
primary_usage="mock_primary_usage",
)
assert session.access_token.get_secret_value() == "mock_access_token"
assert session.token_type == "mock_token_type"
assert session.email == "mock_email"
assert session.user_uuid == "mock_user_uuid"
assert session.username == "mock_username"
assert session.primary_usage == "mock_primary_usage"
def test_fields():
"""Test the HubSession fields."""
fields = HubSession.model_fields
fields_keys = fields.keys()
assert "access_token" in fields_keys
assert "token_type" in fields_keys
assert "email" in fields_keys
assert "user_uuid" in fields_keys
assert "username" in fields_keys
assert "primary_usage" in fields_keys

10
openbb_platform/core/tests/app/model/hub/test_hub_user_settings.py
View File

@@ -1,10 +0,0 @@
"""Test the HubUserSettings class."""
from openbb_core.app.model.hub.hub_user_settings import HubUserSettings
def test_hub_user_settings():
"""Test the HubUserSettings class."""
hub_settings = HubUserSettings()
assert isinstance(hub_settings, HubUserSettings)

9
openbb_platform/core/tests/app/model/test_profile.py
View File

@@ -1,9 +0,0 @@
"""Test the profile class."""
from openbb_core.app.model.profile import Profile
def test_preferences():
"""Test the preferences class."""
preferences = Profile()
assert isinstance(preferences, Profile)

2
openbb_platform/core/tests/app/model/test_user_settings.py
View File

@@ -3,7 +3,6 @@
from openbb_core.app.model.credentials import Credentials
from openbb_core.app.model.defaults import Defaults
from openbb_core.app.model.preferences import Preferences
from openbb_core.app.model.profile import Profile
from openbb_core.app.model.user_settings import UserSettings
@@ -11,7 +10,6 @@ def test_user_settings():
"""Test the UserSettings model."""
settings = UserSettings(
credentials=Credentials(),
profile=Profile(),
preferences=Preferences(),
defaults=Defaults(),
)

397
openbb_platform/core/tests/app/service/test_hub_service.py
View File

@@ -1,397 +0,0 @@
"""Test the hub_service.py module."""
# pylint: disable=W0212
# ruff: noqa: S105 S106
from pathlib import Path
from time import time
from unittest.mock import MagicMock, patch
import pytest
from jwt import encode
from openbb_core.app.model.defaults import Defaults
from openbb_core.app.service.hub_service import (
Credentials,
HubService,
HubSession,
HubUserSettings,
OpenBBError,
)
from pydantic import SecretStr
@pytest.fixture
def mocker():
"""Fixture for mocker."""
with patch("openbb_core.app.service.hub_service.HubService") as mock:
yield mock
def test_v3tov4_map():
"""Test v3 to v4 map."""
v3_keys = {
"databento": "API_DATABENTO_KEY",
"alpha_vantage": "API_KEY_ALPHAVANTAGE",
"fmp": "API_KEY_FINANCIALMODELINGPREP",
"nasdaq": "API_KEY_QUANDL",
"polygon": "API_POLYGON_KEY",
"fred": "API_FRED_KEY",
"news_api": "API_NEWS_TOKEN",
"biztoc": "API_BIZTOC_TOKEN",
"cmc": "API_CMC_KEY",
"finnhub": "API_FINNHUB_KEY",
"whale_alert": "API_WHALE_ALERT_KEY",
"glassnode": "API_GLASSNODE_KEY",
"coinglass": "API_COINGLASS_KEY",
"ethplorer": "API_ETHPLORER_KEY",
"cryptopanic": "API_CRYPTO_PANIC_KEY",
"crypto_panic": "API_CRYPTO_PANIC_KEY", # If dev choses to use this name
"bitquery": "API_BITQUERY_KEY",
"smartstake": ["API_SMARTSTAKE_KEY", "API_SMARTSTAKE_TOKEN"],
"messari": "API_MESSARI_KEY",
"shroom": "API_SHROOM_KEY",
"santiment": "API_SANTIMENT_KEY",
"eodhd": "API_EODHD_KEY",
"tokenterminal": "API_TOKEN_TERMINAL_KEY",
"token_terminal": "API_TOKEN_TERMINAL_KEY", # If dev choses to use this name
"intrinio": "API_INTRINIO_KEY",
"github": "API_GITHUB_KEY",
"reddit": [
"API_REDDIT_CLIENT_ID",
"API_REDDIT_CLIENT_SECRET",
"API_REDDIT_USERNAME",
"API_REDDIT_USER_AGENT",
"API_REDDIT_PASSWORD",
],
"companies_house": "API_COMPANIESHOUSE_KEY",
"companieshouse": "API_COMPANIESHOUSE_KEY", # If dev choses to use this name
"dappradar": "API_DAPPRADAR_KEY",
"nixtla": "API_KEY_NIXTLA",
}
providers = sorted(
[
p.stem
for p in Path("openbb_platform", "providers").glob("*")
if p.is_dir() and p.name not in ("__pycache__", "tests")
]
)
for provider in providers:
if provider in v3_keys:
keys = v3_keys[provider]
keys_list = keys if isinstance(keys, list) else [keys]
for k in keys_list:
assert k.lower() in HubService.V3TOV4
def test_connect_with_email_password():
"""Test connect with email and password."""
mock_hub_session = MagicMock(spec=HubSession)
with (
patch(
"requests.post", return_value=MagicMock(status_code=200, json=lambda: {})
),
patch.object(
HubService,
"_get_session_from_email_password",
return_value=mock_hub_session,
),
):
hub_service = HubService()
result = hub_service.connect(email="test@example.com", password="password")
assert result == mock_hub_session
assert hub_service.session == mock_hub_session
def test_connect_with_sdk_token():
"""Test connect with Platform personal access token."""
mock_hub_session = MagicMock(spec=HubSession)
with (
patch(
"requests.post", return_value=MagicMock(status_code=200, json=lambda: {})
),
patch.object(
HubService,
"_get_session_from_platform_token",
return_value=mock_hub_session,
),
):
hub_service = HubService()
result = hub_service.connect(pat="pat")
assert result == mock_hub_session
assert hub_service.session == mock_hub_session
def test_connect_without_credentials():
"""Test connect without credentials."""
hub_service = HubService()
with pytest.raises(
OpenBBError, match="Please provide 'email' and 'password' or 'pat'"
):
hub_service.connect()
def test_get_session_from_email_password():
"""Test get session from email and password."""
mock_hub_session = MagicMock(spec=HubSession)
with (
patch(
"requests.post",
return_value=MagicMock(
status_code=200,
json=lambda: {
"access_token": "token",
"token_type": "Bearer",
"uuid": "uuid",
"email": "email",
"username": "username",
"primary_usage": "primary_usage",
},
),
),
patch.object(
HubService,
"_get_session_from_email_password",
return_value=mock_hub_session,
),
):
hub_service = HubService()
result = hub_service._get_session_from_email_password("email", "password")
assert isinstance(result, HubSession)
def test_get_session_from_platform_token():
"""Test get session from Platform personal access token."""
mock_hub_session = MagicMock(spec=HubSession)
with (
patch(
"requests.post",
return_value=MagicMock(
status_code=200,
json=lambda: {
"access_token": "token",
"token_type": "Bearer",
"uuid": "uuid",
"username": "username",
"email": "email",
"primary_usage": "primary_usage",
},
),
),
patch.object(
HubService,
"_get_session_from_platform_token",
return_value=mock_hub_session,
),
):
mock_token = (
"eyJ0eXAiOiJKV1QiLCJhbGciOiJFUzI1NiIsImtpZCI6ImRiMjEyZDdhZj"
"c2MWI0ZTNlOGNjZGM3OWQ5Zjk4YWM5In0.eyJhY2Nlc3NfdG9rZW4iOiJ0"
"b2tlbiIsInRva2VuX3R5cGUiOiJCZWFyZXIiLCJ1dWlkIjoidXVpZCIsInV"
"zZXJuYW1lIjoidXNlcm5hbWUiLCJlbWFpbCI6ImVtYWlsIiwicHJpbWFyeV9"
"1c2FnZSI6InByaW1hcnlfdXNhZ2UifQ.FAtE8-a1a-313Zoa6dREIxGZOHaW9"
"-JLZnFzyJ6dlHBZnkjQT2tfaaefxnTdAlSmToQwxGykvuatmI7L0wztPQ"
)
result = HubService()._get_session_from_platform_token(mock_token)
assert isinstance(result, HubSession)
def test_disconnect():
"""Test disconnect."""
with (
patch(
"requests.get",
return_value=MagicMock(
status_code=200,
json=lambda: {"success": True},
),
),
patch.object(
HubService,
"_post_logout",
return_value=True,
),
):
mock_hub_session = MagicMock(
spec=HubSession, access_token=SecretStr("token"), token_type="Bearer"
)
hub_service = HubService(mock_hub_session)
assert hub_service.disconnect() is True
assert hub_service.session is None
def test_get_user_settings():
"""Test get user settings."""
with (
patch(
"requests.get",
return_value=MagicMock(
status_code=200,
json=lambda: {},
),
),
patch.object(
HubService,
"_get_user_settings",
return_value=MagicMock(spec=HubUserSettings),
),
):
mock_hub_session = MagicMock(
spec=HubSession, access_token=SecretStr("token"), token_type="Bearer"
)
hub_service = HubService(mock_hub_session)
user_settings = hub_service._get_user_settings()
assert isinstance(user_settings, HubUserSettings)
def test_put_user_settings():
"""Test put user settings."""
with (
patch(
"requests.put",
return_value=MagicMock(
status_code=200,
),
),
patch.object(
HubService,
"_put_user_settings",
return_value=True,
),
):
mock_hub_session = MagicMock(
spec=HubSession, access_token=SecretStr("token"), token_type="Bearer"
)
mock_user_settings = MagicMock(spec=HubUserSettings)
hub_service = HubService(mock_hub_session)
assert (
hub_service._put_user_settings(mock_hub_session, mock_user_settings) is True
)
def test_hub2platform_v4_only():
"""Test hub2platform."""
mock_user_settings = MagicMock(spec=HubUserSettings)
mock_user_settings.features_keys = {
"fmp_api_key": "abc",
"polygon_api_key": "def",
"fred_api_key": "ghi",
}
mock_user_settings.features_settings = {}
credentials, _ = HubService().hub2platform(mock_user_settings)
assert isinstance(credentials, Credentials)
assert credentials.fmp_api_key.get_secret_value() == "abc"
assert credentials.polygon_api_key.get_secret_value() == "def"
assert credentials.fred_api_key.get_secret_value() == "ghi"
def test_hub2platform_v3_only():
"""Test hub2platform."""
mock_user_settings = MagicMock(spec=HubUserSettings)
mock_user_settings.features_keys = {
"api_key_financialmodelingprep": "abc",
"api_polygon_key": "def",
"api_fred_key": "ghi",
}
mock_user_settings.features_settings = {}
credentials, _ = HubService().hub2platform(mock_user_settings)
assert isinstance(credentials, Credentials)
assert credentials.fmp_api_key.get_secret_value() == "abc"
assert credentials.polygon_api_key.get_secret_value() == "def"
assert credentials.fred_api_key.get_secret_value() == "ghi"
def test_hub2platform_v3v4():
"""Test hub2platform."""
mock_user_settings = MagicMock(spec=HubUserSettings)
mock_user_settings.features_keys = {
"api_key_financialmodelingprep": "abc",
"fmp_api_key": "other_key",
"api_polygon_key": "def",
"api_fred_key": "ghi",
}
mock_user_settings.features_settings = {}
credentials, _ = HubService().hub2platform(mock_user_settings)
assert isinstance(credentials, Credentials)
assert credentials.fmp_api_key.get_secret_value() == "other_key"
assert credentials.polygon_api_key.get_secret_value() == "def"
assert credentials.fred_api_key.get_secret_value() == "ghi"
def test_platform2hub():
"""Test platform2hub."""
mock_user_settings = MagicMock(spec=HubUserSettings)
mock_user_settings.features_keys = { # Received from Hub
"api_key_financialmodelingprep": "abc",
"fmp_api_key": "other_key",
"api_fred_key": "ghi",
}
mock_user_settings.features_settings = {}
mock_hub_service = HubService()
mock_hub_service._hub_user_settings = mock_user_settings
mock_credentials = Credentials( # Current credentials
fmp_api_key=SecretStr("fmp"),
polygon_api_key=SecretStr("polygon"),
fred_api_key=SecretStr("fred"),
benzinga_api_key=SecretStr("benzinga"),
some_api_key=SecretStr("some"),
)
mock_defaults = Defaults()
user_settings = mock_hub_service.platform2hub(mock_credentials, mock_defaults)
assert isinstance(user_settings, HubUserSettings)
assert user_settings.features_keys["api_key_financialmodelingprep"] == "fmp"
assert user_settings.features_keys["fmp_api_key"] == "other_key"
assert user_settings.features_keys["polygon_api_key"] == "polygon"
assert user_settings.features_keys["api_fred_key"] == "fred"
assert user_settings.features_keys["benzinga_api_key"] == "benzinga"
assert "some_api_key" in user_settings.features_keys
assert "defaults" in user_settings.features_settings
@pytest.mark.parametrize(
"offset, message",
[
# valid
(
100,
None,
),
# expired
(
0,
"Platform personal access token expired.",
),
# invalid
(None, "Failed to decode Platform token."),
],
)
def test__check_token_expiration(offset, message):
"""Test check token expiration function."""
token = (
encode(
{"some": "payload", "exp": int(time()) + offset},
"secret",
algorithm="HS256",
)
if offset is not None
else "invalid_token"
)
if message:
with pytest.raises(OpenBBError, match=message):
HubService._check_token_expiration(token)
else:
HubService._check_token_expiration(token)

61
openbb_platform/core/tests/app/static/test_app_factory.py
View File

@@ -2,16 +2,11 @@
# pylint: disable=redefined-outer-name
from unittest.mock import PropertyMock, patch
import pytest
from openbb_core.app.model.hub.hub_session import HubSession
from openbb_core.app.model.system_settings import SystemSettings
from openbb_core.app.model.user_settings import UserSettings
from openbb_core.app.static.account import Account
from openbb_core.app.static.app_factory import create_app
from openbb_core.app.static.coverage import Coverage
from pydantic import SecretStr
@pytest.fixture(scope="module")
@@ -39,62 +34,6 @@ def test_app_user_settings(app_factory):
assert isinstance(user_settings, UserSettings)
def test_app_account(app_factory):
"""Test app account."""
account = app_factory.account
assert account
assert isinstance(account, Account)
# flake8: noqa: S106 # pylint: disable=W0212
@patch("openbb_core.app.static.account.HubService", autospec=True)
def test_app_account_login(mock_hub_service, app_factory):
"""Test app account login."""
mock_hub_service.return_value.pull.return_value = UserSettings()
mock_hub_service.return_value.connect.return_value = None
# Mock the session object
mock_session = HubSession(
access_token=SecretStr("mock_token"),
token_type="bearer",
user_uuid="mock_user_uuid",
email="test@example.com",
primary_usage="mock_usage",
)
# We need to mock the _session attribute and session property on the HubService instance
mock_hub_service.return_value._session = mock_session
type(mock_hub_service.return_value).session = PropertyMock(
return_value=mock_session
)
# Get the account object from the app factory
account = app_factory.account
assert account
assert hasattr(account, "login")
assert callable(account.login)
# Call the login method
account.login(email="test@example.com", password="password", remember_me=True)
# Assert that the hub_session is set in the user settings
assert app_factory.user.profile.hub_session is not None
assert (
app_factory.user.profile.hub_session.access_token.get_secret_value()
== "mock_token"
)
# Assert that the HubService was called correctly
mock_hub_service.return_value.connect.assert_called_once_with(
"test@example.com", "password", None
)
mock_hub_service.return_value.pull.assert_called_once()
# Test logout
with patch("openbb_core.app.static.account.Path.unlink") as mock_unlink:
account.logout()
mock_unlink.assert_called_once()
def test_app_coverage(app_factory):
"""Test app coverage."""
coverage = app_factory.coverage

1
openbb_platform/extensions/platform_api/README.md vendored
View File

@@ -65,7 +65,6 @@ Launcher specific arguments:
--editable Flag to make widgets.json an editable file that can be modified during runtime. Default is 'false'.
--build If the file already exists, changes prompt action to overwrite/append/ignore. Only valid when --editable true.
--no-build Do not build the widgets.json file. Use this flag to load an existing widgets.json file without checking for updates.
--login Login to the OpenBB Platform.
--exclude JSON encoded list of API paths to exclude from widgets.json. Disable entire routes with '*' - e.g. '["/api/v1/*"]'.
--no-filter Do not filter out widgets in widget_settings.json file.
--widgets-json Absolute/relative path to use as the widgets.json file. Default is ~/envs/{env}/assets/widgets.json, when --editable is 'true'.

106
openbb_platform/extensions/platform_api/openbb_platform_api/utils/api.py
View File

@@ -22,7 +22,6 @@ Launcher specific arguments:
--editable Flag to make widgets.json an editable file that can be modified during runtime. Default is 'false'.
--build If the file already exists, changes prompt action to overwrite/append/ignore. Only valid when --editable true.
--no-build Do not build the widgets.json file. Use this flag to load an existing widgets.json file without checking for updates.
--login Login to the OpenBB Platform.
--exclude JSON encoded list of API paths to exclude from widgets.json. Disable entire routes with '*' - e.g. '["/api/v1/*"]'.
--no-filter Do not filter out widgets in widget_settings.json file.
--widgets-json Absolute/relative path to use as the widgets.json file. Default is ~/envs/{env}/assets/widgets.json, when --editable is 'true'.
@@ -91,115 +90,18 @@ def check_port(host, port):
return port
def get_user_settings(
_login: bool, current_user_settings: str, user_settings_copy: str
):
def get_user_settings(current_user_settings: str) -> dict:
"""Login to the OpenBB Platform."""
# pylint: disable=import-outside-toplevel
import getpass
if Path(current_user_settings).exists():
with open(current_user_settings, encoding="utf-8") as f:
_current_settings = json.load(f)
user_settings = json.load(f)
else:
_current_settings = {
user_settings = {
"credentials": {},
"preferences": {},
"defaults": {"commands": {}},
}
if (isinstance(_login, str) and _login.lower() == "false") or not _login:
return _current_settings
pat = getpass.getpass(
"\n\nEnter your personal access token (PAT) to authorize the API and update your local settings."
+ "\nSkip to use a pre-configured 'user_settings.json' file."
+ "\nPress Enter to skip or copy (entered values are not displayed on screen) your PAT to the command line: "
)
if pat:
from openbb_core.app.service.hub_service import HubService
hub_credentials: dict = {}
hub_preferences: dict = {}
hub_defaults: dict = {}
try:
Hub = HubService()
_ = Hub.connect(pat=pat)
hub_settings = Hub.pull()
hub_credentials = json.loads(
hub_settings.credentials.model_dump_json()
) # pylint: disable=no-member
hub_preferences = json.loads(
hub_settings.preferences.model_dump_json()
) # pylint: disable=no-member
hub_defaults = json.loads(
hub_settings.defaults.model_dump_json()
) # pylint: disable=no-member
except Exception as e: # pylint: disable=broad-exception-caught
print( # noqa: T201
f"\n\nError connecting with Hub:\n{e}\n\nUsing the local settings.\n"
)
if hub_credentials:
# Prompt the user to ask if they want to persist the new settings
persist_input = (
input(
"\n\nDo you want to persist the new settings?"
+ " Not recommended for public machines. (yes/no): "
)
.strip()
.lower()
)
if persist_input in ["yes", "y"]:
PERSIST = True
elif persist_input in ["no", "n"]:
PERSIST = False
else:
print( # noqa
"\n\nInvalid input. Defaulting to not persisting the new settings."
)
PERSIST = False
# Save the current settings to restore at the end of the session.
if PERSIST is False:
with open(user_settings_copy, "w", encoding="utf-8") as f:
json.dump(_current_settings, f, indent=4)
new_settings = _current_settings.copy()
new_settings.setdefault("credentials", {})
new_settings.setdefault("preferences", {})
new_settings.setdefault("defaults", {"commands": {}})
# Update the current settings with the new settings
if hub_credentials:
for k, v in hub_credentials.items():
if v:
new_settings["credentials"][k] = v.strip('"').strip("'")
if hub_preferences:
for k, v in hub_preferences.items():
if v:
new_settings["preferences"][k] = v
if hub_defaults:
for k, v in hub_defaults.items():
if k == "commands":
for key, value in hub_defaults["commands"].items():
if value:
new_settings["defaults"]["commands"][key] = value
elif v:
new_settings["defaults"][k] = v
else:
continue
# Write the new settings to the user_settings.json file
with open(current_user_settings, "w", encoding="utf-8") as f:
json.dump(new_settings, f, indent=4)
_current_settings = new_settings
return _current_settings
return user_settings
def get_widgets_json(

8
openbb_platform/extensions/platform_api/openbb_platform_api/utils/openapi.py
View File

@@ -378,11 +378,11 @@ def process_parameter(param: dict, providers: list[str]) -> dict:
# Safe check for description
if (
p.get("description")
and "Multiple comma separated items allowed" in p["description"]
p.get("description", "")
and "Multiple comma separated items allowed" in p["description"] # type: ignore
):
p["description"] = (
p["description"].split("Multiple comma separated items allowed")[0].strip()
p["description"].split("Multiple comma separated items allowed")[0].strip() # type: ignore
)
if x_widget_config := param.get(
@@ -432,7 +432,7 @@ def process_parameter(param: dict, providers: list[str]) -> dict:
p["options"] = {} if providers else []
if providers:
for provider in providers:
p["options"][provider] = []
p["options"][provider] = [] # type: ignore
# Handle widget config
if _widget_config := p_schema.get("x-widget_config", {}):

4
openbb_platform/extensions/platform_api/openbb_platform_api/utils/widgets.py
View File

@@ -408,7 +408,7 @@ def build_json( # noqa: PLR0912 # pylint: disable=too-many-branches, too-many-
openapi, post_body_source.get("operationId"), route, _param
)
modified_post_params = modify_query_schema(
post_params,
post_params, # type: ignore
provider, # type: ignore
)
@@ -456,7 +456,7 @@ def build_json( # noqa: PLR0912 # pylint: disable=too-many-branches, too-many-
form_endpoint_path if has_form_endpoint else route,
)
modified_post_params = modify_query_schema(
post_params,
post_params, # type: ignore
provider, # type: ignore
)

4
openbb_platform/extensions/platform_api/tests/test_api.py
View File

@@ -39,9 +39,7 @@ def test_get_user_settings_no_login():
read_data='{"credentials": {}, "preferences": {}, "defaults": {"commands": {}}}'
),
):
settings = get_user_settings(
_login=False, current_user_settings="", user_settings_copy=""
)
settings = get_user_settings(current_user_settings="")
assert settings == {
"credentials": {},
"preferences": {},