mirror of
https://github.com/OpenBB-finance/OpenBB.git
synced 2026-11-06 09:22:26 +00:00
first version for login prompt
This commit is contained in:
45
openbb_terminal/account/login_controller.py
Normal file
45
openbb_terminal/account/login_controller.py
Normal file
@@ -0,0 +1,45 @@
|
||||
from openbb_terminal.rich_config import console
|
||||
from openbb_terminal.account.login_model import (
|
||||
get_saved_token_if_valid,
|
||||
request_token,
|
||||
launch_terminal,
|
||||
)
|
||||
|
||||
|
||||
def display_welcome_message():
|
||||
console.print(
|
||||
"You need to be connected to use the installer version of OpenBB."
|
||||
"\nTo register please go on: https://my.openbb.co/register"
|
||||
)
|
||||
|
||||
|
||||
def login_prompt():
|
||||
display_welcome_message()
|
||||
while True:
|
||||
console.print("\nPlease enter your credentials:")
|
||||
email = console.input("> Email: ", style="yellow")
|
||||
password = console.getpass("> Password: ", style="yellow")
|
||||
save_str = console.input("> Keep me logged in? [y/n]: ").lower()
|
||||
|
||||
if save_str == "y":
|
||||
save = True
|
||||
elif save_str == "n":
|
||||
save = False
|
||||
|
||||
login_info = request_token(email, password, save)
|
||||
if login_info:
|
||||
break
|
||||
|
||||
return launch_terminal(login_info)
|
||||
|
||||
|
||||
def main():
|
||||
"""Main function"""
|
||||
login_info = get_saved_token_if_valid()
|
||||
if login_info:
|
||||
return launch_terminal(login_info)
|
||||
return login_prompt()
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
104
openbb_terminal/account/login_model.py
Normal file
104
openbb_terminal/account/login_model.py
Normal file
@@ -0,0 +1,104 @@
|
||||
from dataclasses import dataclass
|
||||
import os.path
|
||||
import json
|
||||
import requests
|
||||
from openbb_terminal import terminal_controller
|
||||
from openbb_terminal.core.config.paths import SETTINGS_DIRECTORY
|
||||
from openbb_terminal.rich_config import console
|
||||
from openbb_terminal import feature_flags
|
||||
import jwt
|
||||
|
||||
BASE_URL = "http://127.0.0.1:8000/terminal/"
|
||||
SUCCESS_MSG = "[green]\nLogin successful[/green]"
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class User:
|
||||
email: str
|
||||
|
||||
|
||||
def launch_terminal(login_info: dict):
|
||||
token = login_info.get("access_token", "")
|
||||
if token:
|
||||
decoded_info = jwt.decode(token, options={"verify_signature": False})
|
||||
User.email = decoded_info.get("sub", "")
|
||||
setattr(feature_flags, "USE_FLAIR", User.email)
|
||||
|
||||
terminal_controller.parse_args_and_run()
|
||||
|
||||
|
||||
def request_token(email: str, password: str, save: bool) -> dict:
|
||||
"""Request token and save it to file
|
||||
|
||||
Parameters
|
||||
----------
|
||||
email : str
|
||||
Email
|
||||
password : str
|
||||
Password
|
||||
|
||||
"""
|
||||
data = {
|
||||
"email": email,
|
||||
"password": password,
|
||||
"remember": True,
|
||||
}
|
||||
response = requests.post(BASE_URL + "login", json=data)
|
||||
code = response.status_code
|
||||
if code == 200:
|
||||
console.print(SUCCESS_MSG)
|
||||
login = response.json()
|
||||
if save:
|
||||
with open(SETTINGS_DIRECTORY / "login.json", "w") as outfile:
|
||||
outfile.write(json.dumps(login))
|
||||
return login
|
||||
if code == 401:
|
||||
console.print(f"[red]\n{str(response.json()['detail']).capitalize()}[/red]\n")
|
||||
elif code == 403:
|
||||
console.print(f"[red]\n{str(response.json()['message']).capitalize()}[/red]\n")
|
||||
else:
|
||||
console.print("[red]\nUnknown error.[/red]\n")
|
||||
return {}
|
||||
|
||||
|
||||
def get_saved_token_if_valid() -> dict:
|
||||
"""Try to login with saved credentials
|
||||
|
||||
Returns
|
||||
-------
|
||||
bool
|
||||
True if login was successful, False otherwise
|
||||
"""
|
||||
file_path = SETTINGS_DIRECTORY / "login.json"
|
||||
if os.path.isfile(file_path):
|
||||
with open(file_path, "r") as file:
|
||||
login_info = json.load(file)
|
||||
if is_token_valid(login_info):
|
||||
return login_info
|
||||
return {}
|
||||
|
||||
|
||||
def is_token_valid(login_info: dict) -> bool:
|
||||
"""Request login with saved credentials
|
||||
|
||||
Parameters
|
||||
----------
|
||||
login_info : dict
|
||||
Login information
|
||||
|
||||
Returns
|
||||
-------
|
||||
bool
|
||||
True if login was successful, False otherwise
|
||||
"""
|
||||
if "access_token" in login_info and "token_type" in login_info:
|
||||
response = requests.get(
|
||||
url=BASE_URL + "user",
|
||||
headers={
|
||||
"Authorization": f"{login_info['token_type'].title()} {login_info['access_token']}"
|
||||
},
|
||||
)
|
||||
if response.status_code == 200:
|
||||
console.print(SUCCESS_MSG)
|
||||
return True
|
||||
return False
|
||||
@@ -1,6 +1,7 @@
|
||||
"""Rich Module"""
|
||||
__docformat__ = "numpy"
|
||||
|
||||
import getpass
|
||||
import os
|
||||
import json
|
||||
from pathlib import Path
|
||||
@@ -314,5 +315,13 @@ class ConsoleAndPanel:
|
||||
else:
|
||||
print(*args, **kwargs)
|
||||
|
||||
def input(self, *args, **kwargs):
|
||||
self.print(*args, **kwargs, end="")
|
||||
return input()
|
||||
|
||||
def getpass(self, *args, **kwargs):
|
||||
self.print(*args, **kwargs, end="")
|
||||
return getpass.getpass("")
|
||||
|
||||
|
||||
console = ConsoleAndPanel()
|
||||
|
||||
@@ -2,11 +2,17 @@ import multiprocessing
|
||||
import sys
|
||||
from openbb_terminal import terminal_controller
|
||||
from openbb_terminal.core.integration_tests import integration_controller
|
||||
from openbb_terminal.account import login_controller
|
||||
from openbb_terminal.terminal_helper import is_packaged_application
|
||||
|
||||
IS_PACKAGE = True # is_packaged_application() uncomment to test in installer
|
||||
|
||||
if __name__ == "__main__":
|
||||
multiprocessing.freeze_support()
|
||||
sent_args = sys.argv[1:]
|
||||
if "-t" in sent_args or "--test" in sent_args:
|
||||
integration_controller.main()
|
||||
elif IS_PACKAGE:
|
||||
login_controller.main()
|
||||
else:
|
||||
terminal_controller.parse_args_and_run()
|
||||
|
||||
Reference in New Issue
Block a user