mirror of
https://github.com/OpenBB-finance/OpenBB.git
synced 2026-10-29 13:32:08 +00:00
change controller
This commit is contained in:
@@ -1,14 +1,14 @@
|
||||
from typing import Tuple
|
||||
from openbb_terminal.core.config.paths import PACKAGE_DIRECTORY
|
||||
from openbb_terminal.rich_config import console
|
||||
import json
|
||||
from openbb_terminal import terminal_controller
|
||||
from openbb_terminal.account.login_model import (
|
||||
apply_configs,
|
||||
fetch_user_configs,
|
||||
get_login_info,
|
||||
load_user_info,
|
||||
request_login_info,
|
||||
)
|
||||
from openbb_terminal.account.sync_model import fetch_user_configs, apply_configs
|
||||
|
||||
|
||||
def display_welcome_message():
|
||||
@@ -57,11 +57,24 @@ def login_prompt(welcome=True):
|
||||
if isinstance(login_info, dict) and login_info:
|
||||
break
|
||||
|
||||
# Login successful
|
||||
login(login_info)
|
||||
|
||||
|
||||
def login(login_info: dict):
|
||||
"""Login and launch terminal.
|
||||
|
||||
Parameters
|
||||
----------
|
||||
login_info : dict
|
||||
The login information.
|
||||
"""
|
||||
load_user_info(login_info)
|
||||
fetch_user_configs()
|
||||
apply_configs()
|
||||
terminal_controller.parse_args_and_run()
|
||||
response = fetch_user_configs(login_info)
|
||||
if response.status_code == 200:
|
||||
apply_configs(configs=json.loads(response.content))
|
||||
terminal_controller.parse_args_and_run()
|
||||
else:
|
||||
login_prompt(welcome=False)
|
||||
|
||||
|
||||
def main():
|
||||
@@ -70,13 +83,7 @@ def main():
|
||||
if not login_info:
|
||||
login_prompt()
|
||||
else:
|
||||
load_user_info(login_info)
|
||||
status = fetch_user_configs()
|
||||
if status == 200:
|
||||
apply_configs()
|
||||
terminal_controller.parse_args_and_run()
|
||||
else:
|
||||
login_prompt(welcome=False)
|
||||
login(login_info)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
|
||||
@@ -4,52 +4,11 @@ import json
|
||||
from typing import Union
|
||||
import jwt
|
||||
import requests
|
||||
from openbb_terminal.base_helpers import strtobool
|
||||
from openbb_terminal.core.config.paths import SETTINGS_DIRECTORY
|
||||
from openbb_terminal.rich_config import console
|
||||
from openbb_terminal import feature_flags as obbff
|
||||
from openbb_terminal.account.statics import BASE_URL, Failure
|
||||
from openbb_terminal.account import user
|
||||
from openbb_terminal import config_terminal as cfg
|
||||
|
||||
|
||||
def fetch_user_configs() -> Union[int, Failure]:
|
||||
try:
|
||||
response = requests.get(
|
||||
url=BASE_URL + "terminal/user",
|
||||
headers={"Authorization": f"{user.TOKEN_TYPE.title()} {user.TOKEN}"},
|
||||
)
|
||||
if response.status_code == 200:
|
||||
user.CONFIGS = json.loads(response.content)
|
||||
console.print("[green]\nFetched user configurations.[/green]")
|
||||
return response.status_code
|
||||
console.print("[red]\nFailed to fetch configurations.[/red]")
|
||||
return response.status_code
|
||||
except requests.exceptions.ConnectionError:
|
||||
return Failure("[red]\nConnection error.[/red]")
|
||||
except Exception:
|
||||
return Failure("[red]\nFailed to fetch configurations.[/red]")
|
||||
|
||||
|
||||
def apply_configs():
|
||||
"""Apply configurations."""
|
||||
console.print(user.CONFIGS, style="red")
|
||||
if user.CONFIGS:
|
||||
settings = user.CONFIGS.get("features_settings", {})
|
||||
for k, v in settings.items():
|
||||
if hasattr(obbff, k):
|
||||
if isinstance(getattr(obbff, k), int):
|
||||
setattr(obbff, k, strtobool(v))
|
||||
else:
|
||||
setattr(obbff, k, v)
|
||||
|
||||
keys = user.CONFIGS.get("features_keys", {})
|
||||
for k, v in keys.items():
|
||||
if hasattr(cfg, k):
|
||||
if isinstance(getattr(cfg, k), int):
|
||||
setattr(cfg, k, strtobool(v))
|
||||
else:
|
||||
setattr(cfg, k, v)
|
||||
|
||||
|
||||
def load_user_info(login_info: dict):
|
||||
|
||||
@@ -3,9 +3,8 @@ from typing import Union
|
||||
import requests
|
||||
from openbb_terminal.core.config.paths import SETTINGS_DIRECTORY
|
||||
from openbb_terminal.helper_funcs import system_clear
|
||||
from openbb_terminal.rich_config import console
|
||||
from openbb_terminal.account.statics import BASE_URL, Success, Failure
|
||||
from openbb_terminal.account import login_controller, user
|
||||
from openbb_terminal.account import user
|
||||
|
||||
|
||||
def remote_logout() -> Union[Success, Failure]:
|
||||
@@ -45,7 +44,7 @@ def remove_login_file() -> Union[Success, Failure]:
|
||||
file_path = SETTINGS_DIRECTORY / "login.json"
|
||||
if os.path.isfile(file_path):
|
||||
os.remove(file_path)
|
||||
return Success("[green]\nRemoved login info.[/green]")
|
||||
return Success("[green]\nRemoved login info.[/green]")
|
||||
except Exception:
|
||||
return Failure("[red]\nFailed to remove login file.[/red]")
|
||||
|
||||
|
||||
77
openbb_terminal/account/sync_model.py
Normal file
77
openbb_terminal/account/sync_model.py
Normal file
@@ -0,0 +1,77 @@
|
||||
from typing import Union
|
||||
import requests
|
||||
from openbb_terminal.base_helpers import strtobool
|
||||
from openbb_terminal.rich_config import console
|
||||
from openbb_terminal import feature_flags as obbff
|
||||
from openbb_terminal.account.statics import BASE_URL, Failure
|
||||
from openbb_terminal.account import user
|
||||
from openbb_terminal import config_terminal as cfg
|
||||
|
||||
|
||||
def fetch_user_configs(login_info: dict) -> Union[requests.Response, Failure]:
|
||||
"""Fetch user configurations."""
|
||||
|
||||
token_type = login_info.get("token_type", "")
|
||||
token = login_info.get("access_token", "")
|
||||
|
||||
try:
|
||||
response = requests.get(
|
||||
url=BASE_URL + "terminal/user",
|
||||
headers={"Authorization": f"{token_type.title()} {token}"},
|
||||
)
|
||||
if response.status_code == 200:
|
||||
console.print("[green]\nFetched user configurations.[/green]")
|
||||
else:
|
||||
console.print("[red]\nFailed to fetch configurations.[/red]")
|
||||
return response
|
||||
except requests.exceptions.ConnectionError:
|
||||
return Failure("[red]\nConnection error.[/red]")
|
||||
except Exception:
|
||||
return Failure("[red]\nFailed to fetch configurations.[/red]")
|
||||
|
||||
|
||||
def apply_configs(configs: dict):
|
||||
"""Apply configurations."""
|
||||
console.print(configs, style="red")
|
||||
if configs:
|
||||
settings = configs.get("features_settings", {})
|
||||
for k, v in settings.items():
|
||||
if hasattr(obbff, k):
|
||||
if isinstance(getattr(obbff, k), int):
|
||||
setattr(obbff, k, strtobool(v))
|
||||
else:
|
||||
setattr(obbff, k, v)
|
||||
|
||||
keys = configs.get("features_keys", {})
|
||||
for k, v in keys.items():
|
||||
if hasattr(cfg, k):
|
||||
if isinstance(getattr(cfg, k), int):
|
||||
setattr(cfg, k, strtobool(v))
|
||||
else:
|
||||
setattr(cfg, k, v)
|
||||
|
||||
|
||||
def patch_user_configs(data: dict) -> Union[requests.Response, Failure]:
|
||||
"""Patch user configurations."""
|
||||
# When available on the server just need to send the changed configs, not all of them.
|
||||
# Adapt this to Colin implementation.
|
||||
pass
|
||||
|
||||
|
||||
def put_user_configs(data: dict) -> Union[int, Failure]:
|
||||
"""Push user configurations to the server."""
|
||||
try:
|
||||
response = requests.put(
|
||||
url=BASE_URL + "terminal/user",
|
||||
headers={"Authorization": f"{user.TOKEN_TYPE.title()} {user.TOKEN}"},
|
||||
json=data,
|
||||
)
|
||||
if response.status_code == 200:
|
||||
console.print("[green]\nPushed user configurations.[/green]")
|
||||
else:
|
||||
console.print("[red]\nFailed to push configurations.[/red]")
|
||||
return response.status_code
|
||||
except requests.exceptions.ConnectionError:
|
||||
return Failure("[red]\nConnection error.[/red]")
|
||||
except Exception:
|
||||
return Failure("[red]\nFailed to push configurations.[/red]")
|
||||
@@ -1,7 +1,6 @@
|
||||
from typing import Optional
|
||||
# User information
|
||||
|
||||
TOKEN_TYPE: str = ""
|
||||
TOKEN: str = ""
|
||||
EMAIL: str = ""
|
||||
UUID: str = ""
|
||||
CONFIGS: Optional[dict] = None
|
||||
|
||||
Reference in New Issue
Block a user