mirror of
https://github.com/OpenBB-finance/OpenBB.git
synced 2026-11-01 23:12:21 +00:00
move logout and change to results
This commit is contained in:
@@ -2,13 +2,12 @@ from typing import Tuple
|
||||
from openbb_terminal.core.config.paths import PACKAGE_DIRECTORY
|
||||
from openbb_terminal.rich_config import console
|
||||
from openbb_terminal.account.login_model import (
|
||||
Status,
|
||||
Error,
|
||||
get_login_info,
|
||||
get_login_status,
|
||||
request_login_info,
|
||||
launch_terminal,
|
||||
)
|
||||
from openbb_terminal.account.statics import Success, Failure
|
||||
|
||||
|
||||
def display_welcome_message():
|
||||
@@ -63,13 +62,13 @@ def login_prompt(welcome=True):
|
||||
def main():
|
||||
"""Main function"""
|
||||
login_info = get_login_info()
|
||||
if login_info == Error.UNKNOWN_ERROR or not login_info:
|
||||
if login_info == Failure.UNKNOWN_ERROR or not login_info:
|
||||
login_prompt()
|
||||
else:
|
||||
status = get_login_status(login_info=login_info)
|
||||
if status == Status.VALID_LOGIN:
|
||||
if status == Success.VALID_LOGIN:
|
||||
launch_terminal(login_info=login_info)
|
||||
if status == Status.INVALID_LOGIN:
|
||||
if status == Failure.INVALID_LOGIN:
|
||||
login_prompt(welcome=False)
|
||||
|
||||
|
||||
|
||||
@@ -1,4 +1,3 @@
|
||||
from enum import Enum
|
||||
from pathlib import Path
|
||||
import os.path
|
||||
import json
|
||||
@@ -9,24 +8,9 @@ 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
|
||||
|
||||
from openbb_terminal.account.statics import BASE_URL, Success, Failure
|
||||
from openbb_terminal.account.user import User
|
||||
|
||||
BASE_URL = "http://127.0.0.1:8000/terminal/"
|
||||
|
||||
|
||||
class Status(Enum):
|
||||
VALID_LOGIN = "[green]\nLogin successful.[/green]"
|
||||
INVALID_LOGIN = "[red]\nInvalid login.[/red]"
|
||||
LOGGED_OUT = "[green]\nLogged out.[/green]"
|
||||
|
||||
|
||||
class Error(Enum):
|
||||
CONNECTION_ERROR = "[red]\nConnection error.[/red]"
|
||||
WRONG_CREDENTIALS = "[red]\nWrong credentials.[/red]"
|
||||
UNVERIFIED_EMAIL = "[red]\nUnverified email.[/red]"
|
||||
UNKNOWN_ERROR = "[red]\nUnknown error.[/red]"
|
||||
|
||||
|
||||
def launch_terminal(login_info: dict):
|
||||
"""Launch the terminal.
|
||||
@@ -50,8 +34,8 @@ def launch_terminal(login_info: dict):
|
||||
terminal_controller.parse_args_and_run()
|
||||
|
||||
|
||||
def write_to_file(data: dict, file_path: Path):
|
||||
"""Write data to a file.
|
||||
def save_login_info(data: dict, file_path: Path):
|
||||
"""Save the login info to a file.
|
||||
|
||||
Parameters
|
||||
----------
|
||||
@@ -64,10 +48,10 @@ def write_to_file(data: dict, file_path: Path):
|
||||
with open(file_path, "w") as file:
|
||||
file.write(json.dumps(data))
|
||||
except Exception:
|
||||
console.print(Error.UNKNOWN_ERROR.value)
|
||||
console.print("Failed to save login info.", style="red")
|
||||
|
||||
|
||||
def process_response(response: requests.Response, save: bool) -> Union[dict, Error]:
|
||||
def process_response(response: requests.Response, save: bool) -> Union[dict, Failure]:
|
||||
"""Process the response from the server.
|
||||
|
||||
Parameters
|
||||
@@ -79,26 +63,26 @@ def process_response(response: requests.Response, save: bool) -> Union[dict, Err
|
||||
|
||||
Returns
|
||||
-------
|
||||
Union[dict, Error]
|
||||
Union[dict, Failure]
|
||||
The login info, or an error.
|
||||
"""
|
||||
if response.status_code == 200:
|
||||
console.print(Status.VALID_LOGIN.value)
|
||||
console.print(Success.VALID_LOGIN.value)
|
||||
login = response.json()
|
||||
if save:
|
||||
write_to_file(login, SETTINGS_DIRECTORY / "login.json")
|
||||
save_login_info(login, SETTINGS_DIRECTORY / "login.json")
|
||||
return login
|
||||
if response.status_code == 401:
|
||||
console.print(Error.WRONG_CREDENTIALS.value)
|
||||
return Error.WRONG_CREDENTIALS
|
||||
console.print(Failure.WRONG_CREDENTIALS.value)
|
||||
return Failure.WRONG_CREDENTIALS
|
||||
if response.status_code == 403:
|
||||
console.print(Error.UNVERIFIED_EMAIL.value)
|
||||
return Error.UNVERIFIED_EMAIL
|
||||
console.print(Error.UNKNOWN_ERROR.value)
|
||||
return Error.UNKNOWN_ERROR
|
||||
console.print(Failure.UNVERIFIED_EMAIL.value)
|
||||
return Failure.UNVERIFIED_EMAIL
|
||||
console.print("Failed to login.", style="red")
|
||||
return Failure.UNKNOWN_ERROR
|
||||
|
||||
|
||||
def request_login_info(email: str, password: str, save: bool) -> Union[dict, Error]:
|
||||
def request_login_info(email: str, password: str, save: bool) -> Union[dict, Failure]:
|
||||
"""Request login info from the server.
|
||||
|
||||
Parameters
|
||||
@@ -112,7 +96,7 @@ def request_login_info(email: str, password: str, save: bool) -> Union[dict, Err
|
||||
|
||||
Returns
|
||||
-------
|
||||
Union[dict, Error]
|
||||
Union[dict, Failure]
|
||||
The login info, or an error.
|
||||
"""
|
||||
data = {
|
||||
@@ -124,15 +108,15 @@ def request_login_info(email: str, password: str, save: bool) -> Union[dict, Err
|
||||
try:
|
||||
response = requests.post(BASE_URL + "login", json=data)
|
||||
except requests.exceptions.ConnectionError:
|
||||
console.print(Error.CONNECTION_ERROR.value)
|
||||
return Error.CONNECTION_ERROR
|
||||
console.print(Failure.CONNECTION_ERROR.value)
|
||||
return Failure.CONNECTION_ERROR
|
||||
except Exception:
|
||||
console.print(Error.UNKNOWN_ERROR.value)
|
||||
return Error.UNKNOWN_ERROR
|
||||
console.print("Failed to request login info.", style="red")
|
||||
return Failure.UNKNOWN_ERROR
|
||||
return process_response(response, save)
|
||||
|
||||
|
||||
def get_login_info() -> Union[dict, Error]:
|
||||
def get_login_info() -> Union[dict, Failure]:
|
||||
"""Get the login info from the file.
|
||||
|
||||
Returns
|
||||
@@ -147,12 +131,12 @@ def get_login_info() -> Union[dict, Error]:
|
||||
with open(file_path) as file:
|
||||
return json.load(file)
|
||||
except Exception:
|
||||
console.print(Error.UNKNOWN_ERROR.value)
|
||||
return Error.UNKNOWN_ERROR
|
||||
console.print("Failed to get login info.", style="red")
|
||||
return Failure.UNKNOWN_ERROR
|
||||
return {}
|
||||
|
||||
|
||||
def get_login_status(login_info: dict) -> Union[Status, Error]:
|
||||
def get_login_status(login_info: dict) -> Union[Success, Failure]:
|
||||
"""Check if the login info is valid.
|
||||
|
||||
Parameters
|
||||
@@ -162,53 +146,27 @@ def get_login_status(login_info: dict) -> Union[Status, Error]:
|
||||
|
||||
Returns
|
||||
-------
|
||||
Union[Status, Error]
|
||||
Union[Success, Failure]
|
||||
The status of the login info, or an error.
|
||||
"""
|
||||
if "access_token" in login_info and "token_type" in login_info:
|
||||
try:
|
||||
if (
|
||||
requests.get(
|
||||
url=BASE_URL + "user",
|
||||
url=BASE_URL + "terminal/user",
|
||||
headers={
|
||||
"Authorization": f"{login_info['token_type'].title()} {login_info['access_token']}"
|
||||
},
|
||||
).status_code
|
||||
== 200
|
||||
):
|
||||
console.print(Status.VALID_LOGIN.value)
|
||||
return Status.VALID_LOGIN
|
||||
console.print(Success.VALID_LOGIN.value)
|
||||
return Success.VALID_LOGIN
|
||||
except requests.exceptions.ConnectionError:
|
||||
console.print(Error.CONNECTION_ERROR.value)
|
||||
return Error.CONNECTION_ERROR
|
||||
console.print(Failure.CONNECTION_ERROR.value)
|
||||
return Failure.CONNECTION_ERROR
|
||||
except Exception:
|
||||
console.print(Error.UNKNOWN_ERROR.value)
|
||||
return Error.UNKNOWN_ERROR
|
||||
return Status.INVALID_LOGIN
|
||||
console.print("Failed to get login status.", style="red")
|
||||
return Failure.UNKNOWN_ERROR
|
||||
|
||||
|
||||
def logout() -> Union[Status, Error]:
|
||||
"""Logout the user.
|
||||
|
||||
Returns
|
||||
-------
|
||||
Union[Status, Error]
|
||||
The status of the logout, or an error.
|
||||
"""
|
||||
try:
|
||||
# TODO: Uncomment when the server supports logout
|
||||
# requests.post(
|
||||
# url=BASE_URL + "logout",
|
||||
# headers={"Authorization": f"{User.token_type.title()} {User.token}"},
|
||||
# )
|
||||
file_path = SETTINGS_DIRECTORY / "login.json"
|
||||
if os.path.isfile(file_path):
|
||||
os.remove(file_path)
|
||||
console.print(Status.LOGGED_OUT.value)
|
||||
return Status.LOGGED_OUT
|
||||
except requests.exceptions.ConnectionError:
|
||||
console.print(Error.CONNECTION_ERROR.value)
|
||||
return Error.CONNECTION_ERROR
|
||||
except Exception:
|
||||
console.print(Error.UNKNOWN_ERROR.value)
|
||||
return Error.UNKNOWN_ERROR
|
||||
return Failure.INVALID_LOGIN
|
||||
|
||||
72
openbb_terminal/account/logout_model.py
Normal file
72
openbb_terminal/account/logout_model.py
Normal file
@@ -0,0 +1,72 @@
|
||||
import os.path
|
||||
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.user import User
|
||||
from openbb_terminal.account import login_controller
|
||||
|
||||
|
||||
def remote_logout() -> Union[Success, Failure]:
|
||||
"""Request a remote logout.
|
||||
|
||||
Returns
|
||||
-------
|
||||
Union[Success, Failure]
|
||||
The status of the token invalidation, or an error.
|
||||
"""
|
||||
try:
|
||||
if (
|
||||
requests.get(
|
||||
url=BASE_URL + "logout-everywhere",
|
||||
headers={"Authorization": f"{User.token_type.title()} {User.token}"},
|
||||
).status_code
|
||||
== 200
|
||||
):
|
||||
console.print(Success.REMOTE_LOGOUT.value)
|
||||
return Success.REMOTE_LOGOUT
|
||||
console.print("Failed to logout remotely.", style="red")
|
||||
return Failure.UNKNOWN_ERROR
|
||||
except requests.exceptions.ConnectionError:
|
||||
console.print(Failure.CONNECTION_ERROR.value)
|
||||
return Failure.CONNECTION_ERROR
|
||||
except Exception:
|
||||
console.print("Failed to logout remotely.", style="red")
|
||||
return Failure.UNKNOWN_ERROR
|
||||
|
||||
|
||||
def remove_login_file() -> Union[Success, Failure]:
|
||||
"""Remove the login file.
|
||||
|
||||
Returns
|
||||
-------
|
||||
Success
|
||||
The status of the local logout.
|
||||
"""
|
||||
|
||||
try:
|
||||
file_path = SETTINGS_DIRECTORY / "login.json"
|
||||
if os.path.isfile(file_path):
|
||||
os.remove(file_path)
|
||||
console.print(Success.LOCAL_LOGOUT.value)
|
||||
return Success.LOCAL_LOGOUT
|
||||
except Exception:
|
||||
console.print("Failed to remove login file.", style="red")
|
||||
return Failure.UNKNOWN_ERROR
|
||||
|
||||
|
||||
def logout():
|
||||
"""Logout the user.
|
||||
|
||||
Returns
|
||||
-------
|
||||
Union[Success, Failure]
|
||||
The status of the logout, or an error.
|
||||
"""
|
||||
system_clear()
|
||||
remove_login_file()
|
||||
remote_logout()
|
||||
console.print("")
|
||||
login_controller.login_prompt(welcome=True)
|
||||
26
openbb_terminal/account/statics.py
Normal file
26
openbb_terminal/account/statics.py
Normal file
@@ -0,0 +1,26 @@
|
||||
from enum import Enum
|
||||
from dataclasses import dataclass
|
||||
|
||||
BASE_URL = "http://127.0.0.1:8000/"
|
||||
|
||||
|
||||
class Success(Enum):
|
||||
VALID_LOGIN = "[green]\nLogin successful.[/green]"
|
||||
REMOTE_LOGOUT = "[green]\nLogged out remotely.[/green]"
|
||||
LOCAL_LOGOUT = "[green]\nRemoved login info.[/green]"
|
||||
|
||||
|
||||
class Failure(Enum):
|
||||
INVALID_LOGIN = "[red]\nInvalid login.[/red]"
|
||||
CONNECTION_ERROR = "[red]\nConnection error.[/red]"
|
||||
WRONG_CREDENTIALS = "[red]\nWrong credentials.[/red]"
|
||||
UNVERIFIED_EMAIL = "[red]\nUnverified email.[/red]"
|
||||
UNKNOWN_ERROR = "[red]\nUnknown error.[/red]"
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class User:
|
||||
token_type: str = ""
|
||||
token: str = ""
|
||||
email: str = ""
|
||||
uuid: str = ""
|
||||
@@ -18,7 +18,7 @@ import pandas as pd
|
||||
from prompt_toolkit.formatted_text import HTML
|
||||
from prompt_toolkit.styles import Style
|
||||
from rich.markdown import Markdown
|
||||
from openbb_terminal.account.login_model import Status, logout
|
||||
from openbb_terminal.account.logout_model import logout
|
||||
from openbb_terminal.account.user import User
|
||||
|
||||
from openbb_terminal.core.config.paths import (
|
||||
@@ -49,7 +49,6 @@ from openbb_terminal.stocks import stocks_helper
|
||||
from openbb_terminal.terminal_helper import open_openbb_documentation
|
||||
from openbb_terminal.cryptocurrency import cryptocurrency_helpers
|
||||
from openbb_terminal.core.completer.choices import build_controller_choice_map
|
||||
from openbb_terminal.account import login_controller
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
@@ -673,9 +672,7 @@ class BaseController(metaclass=ABCMeta):
|
||||
ns_parser = self.parse_simple_args(parser, other_args)
|
||||
|
||||
if ns_parser:
|
||||
if logout() == Status.LOGGED_OUT:
|
||||
system_clear()
|
||||
login_controller.main()
|
||||
logout()
|
||||
|
||||
@log_start_end(log=logger)
|
||||
def call_whoami(self, other_args: List[str]) -> None:
|
||||
|
||||
Reference in New Issue
Block a user