feat: add profile-based config management (#112)

Signed-off-by: Panos Vagenas <35837085+vagenas@users.noreply.github.com>

deepsearch/artifacts/cli/main.py

@@ -5,6 +5,7 @@
     DFLT_ARTFCT_INDEX_DIR,
     ArtifactManager,
 )
+from deepsearch.core.cli.utils import cli_handler
 
 app = typer.Typer(no_args_is_help=True, add_completion=False)
 
@@ -31,6 +32,7 @@
 
 
 @app.command(help="List artifacts in index")
+@cli_handler()
 def list_index(
     index: str = INDEX_OPTION,
 ):
@@ -41,6 +43,7 @@ def list_index(
 
 
 @app.command(help="List artifacts in cache")
+@cli_handler()
 def list_cache(
     cache: str = CACHE_OPTION,
 ):
@@ -51,13 +54,15 @@ def list_cache(
 
 
 @app.command(help="Show cache path")
+@cli_handler()
 def locate_default_cache():
     artf_mgr = ArtifactManager()
     path_str = str(artf_mgr.get_cache_path().resolve())
     typer.echo(path_str)
 
 
 @app.command(help="Show path of a cached artifact")
+@cli_handler()
 def locate_cached_artifact(
     artifact_name: str,
     cache: str = CACHE_OPTION,
@@ -69,6 +74,7 @@ def locate_cached_artifact(
 
 
 @app.command(help="Download an artifact to cache")
+@cli_handler()
 def download(
     artifact_name: str,
     index: str = INDEX_OPTION,
@@ -87,6 +93,7 @@ def download(
 
 
 @app.command(help="Download all artifacts to cache")
+@cli_handler()
 def download_all(
     index: str = INDEX_OPTION,
     cache: str = CACHE_OPTION,

deepsearch/core/cli/login.py

@@ -1,47 +1,18 @@
-from pathlib import Path
-
 import typer
 
-from deepsearch.core.client import DeepSearchConfig, DeepSearchKeyAuth
-from deepsearch.core.util.config_paths import config_file_path
+from deepsearch.core.cli.profile_utils import MSG_LOGIN_DEPRECATION
 
 app = typer.Typer(invoke_without_command=True)
 
 
-@app.callback(help="Save authentication configuration")
+@app.callback(help=MSG_LOGIN_DEPRECATION)
 def save_auth(
     *,
-    host: str = typer.Option("https://deepsearch-experience.res.ibm.com", prompt=True),
-    email: str = typer.Option(..., prompt=True),
-    api_key: str = typer.Option(..., prompt=True, hide_input=True),
-    verify_ssl: bool = typer.Option(True, prompt=True),
-    output: str = typer.Option(
-        str(config_file_path()),
-        help="Where to save configuration to. Use '-' to print to stdout",
-    ),
+    host: str = typer.Option(""),
+    email: str = typer.Option(""),
+    api_key: str = typer.Option("", hide_input=True),
+    verify_ssl: bool = typer.Option(True),
+    output: str = typer.Option(""),
 ):
-
-    config = DeepSearchConfig(
-        host=host,
-        auth=DeepSearchKeyAuth(username=email, api_key=api_key),
-        verify_ssl=verify_ssl,
-    )
-
-    contents = config.json(indent=2)
-
-    if output == "-":
-        typer.echo(contents)
-        return
-
-    output_file = Path(output)
-
-    if not output_file.is_file():
-        output_file.parent.mkdir(parents=True, exist_ok=True)
-
-    output_file.write_text(contents, encoding="utf-8")
-
-    typer.secho(f"File {output_file} updated!", fg=typer.colors.GREEN)
-
-
-if __name__ == "__main__":
-    app()
+    print(MSG_LOGIN_DEPRECATION)
+    raise typer.Exit(code=1)

deepsearch/core/cli/main.py

@@ -1,13 +1,18 @@
 import typer
 
 import deepsearch as ds
+from deepsearch.core.cli.profile_utils import MSG_LOGIN_DEPRECATION
 
-from .config import app as config_app
 from .login import app as login_app
+from .profile import app as profile_app
 
-app = typer.Typer(no_args_is_help=True, add_completion=False)
-app.add_typer(config_app, name="config", help="Manage CLI config files")
-app.add_typer(login_app, name="login", help="Login to DeepSearch platform")
+app = typer.Typer(
+    no_args_is_help=True,
+    add_completion=False,
+    pretty_exceptions_enable=False,
+)
+app.add_typer(profile_app, name="profile", help="Manage profile configuration")
+app.add_typer(login_app, name="login", help=MSG_LOGIN_DEPRECATION)
 
 
 @app.command(name="version", help=f"Print the client and server version")

deepsearch/core/cli/profile.py

@@ -0,0 +1,132 @@
+from typing import Optional
+
+import typer
+from rich.console import Console
+from rich.table import Table
+from typing_extensions import Annotated
+
+from deepsearch.core.cli.profile_utils import (
+    MSG_NO_PROFILE_SELECTED,
+    MSG_NO_PROFILES_DEFINED,
+)
+from deepsearch.core.cli.utils import cli_handler
+from deepsearch.core.client.settings import ProfileSettings
+from deepsearch.core.client.settings_manager import settings_mgr
+
+app = typer.Typer(no_args_is_help=True)
+
+console = Console()
+err_console = Console(stderr=True)
+
+
+@app.command(
+    name="config",
+    help=f"Add or update a profile.",
+)
+@cli_handler()
+def add_profile(
+    host: str = typer.Option(prompt=True),
+    username: str = typer.Option(prompt=True),
+    api_key: str = typer.Option(prompt=True, hide_input=True),
+    verify_ssl: bool = typer.Option(default=True),
+    profile_name: str = typer.Option(
+        default="",
+        help="If not set, the active profile will be updated or, if no profile available, a new profile with a predetermined name will be created.",
+    ),
+    activate_profile: bool = typer.Option(default=True),
+):
+    prfl_name = (
+        profile_name if profile_name else settings_mgr.get_profile_name_suggestion()
+    )
+
+    profile_settings = ProfileSettings(
+        host=host,
+        username=username,
+        api_key=api_key,
+        verify_ssl=verify_ssl,
+    )
+
+    settings_mgr.save_settings(
+        profile_settgs=profile_settings,
+        profile_name=prfl_name,
+        activate_profile=activate_profile,
+    )
+
+
+@app.command(
+    name="list",
+    help=f"List all profiles.",
+)
+@cli_handler()
+def list_profiles() -> None:
+    table = Table(
+        "active",
+        "profile",
+    )
+    profiles = settings_mgr.get_all_profile_settings()
+    active_profile = settings_mgr.get_active_profile()
+
+    if len(profiles) > 0:
+        for k in profiles:
+            mark = "*" if k == active_profile else " "
+            table.add_row(
+                mark,
+                k,
+            )
+        console.print(table)
+
+        if active_profile is None:
+            console.print(MSG_NO_PROFILE_SELECTED)
+    else:
+        console.print(MSG_NO_PROFILES_DEFINED)
+
+
+@app.command(
+    name="show",
+    help=f"Display a profile.",
+)
+@cli_handler()
+def show_profile(
+    profile_name: Annotated[
+        Optional[str],
+        typer.Argument(
+            help="If not set, the active profile will be displayed.",
+        ),
+    ] = None
+) -> None:
+    table = Table(
+        "profile",
+        "config",
+    )
+    prfl_name = profile_name or settings_mgr.get_active_profile()
+    profile = settings_mgr.get_profile_settings(profile_name=prfl_name)
+
+    table.add_row(
+        prfl_name,
+        repr(profile.dict()),
+    )
+    console.print(table)
+
+
+@app.command(
+    name="use",
+    help=f"Activate a profile.",
+    no_args_is_help=True,
+)
+@cli_handler()
+def set_default_profile(
+    profile_name: str,
+) -> None:
+    settings_mgr.activate_profile(profile_name=profile_name)
+
+
+@app.command(
+    name="remove",
+    help=f"Remove a profile.",
+    no_args_is_help=True,
+)
+@cli_handler()
+def remove_profile(
+    profile_name: str,
+) -> None:
+    settings_mgr.remove_profile(profile_name=profile_name)

deepsearch/core/cli/profile_utils.py

@@ -0,0 +1,8 @@
+MSG_NO_PROFILE_SELECTED = (
+    "No profile activated; to select one check `deepsearch profile use --help`"
+)
+MSG_NO_PROFILES_DEFINED = (
+    "No profiles defined; to create one check `deepsearch profile config --help`"
+)
+MSG_AMBIGUOUS_SUCCESSOR = "Cannot remove active profile with ambiguous successor; please switch to another profile (check `deepsearch profile use --help`) and try again"
+MSG_LOGIN_DEPRECATION = "Authentication is now done via profiles; to set up one check `deepsearch profile config --help`"

deepsearch/core/cli/utils.py

@@ -0,0 +1,25 @@
+from functools import wraps
+
+import typer
+
+from deepsearch.core.client.settings_manager import settings_mgr
+
+
+def cli_handler():
+    """Decorator for wrapping CLI commands and handling exceptions as controlled via settings"""
+
+    def decorate(func):
+        @wraps(func)
+        def wrap(*args, **kwargs):
+            try:
+                return func(*args, **kwargs)
+            except Exception as e:
+                if settings_mgr.get_show_cli_stack_traces():
+                    raise e
+                else:
+                    typer.secho(str(e), fg=typer.colors.RED)
+                    raise typer.Exit(code=1)
+
+        return wrap
+
+    return decorate

deepsearch/core/client/settings.py

@@ -0,0 +1,61 @@
+from __future__ import annotations
+
+from getpass import getpass
+from pathlib import Path
+from typing import Dict, Optional, Union
+
+from pydantic import BaseSettings, SecretStr
+
+
+class DumpableSettings(BaseSettings):
+    @classmethod
+    def get_env_var_name(cls, attr_name) -> str:
+        return cls.Config.env_prefix + attr_name.upper()
+
+    def _get_serializable_dict(self) -> Dict[str, str]:
+        result = {}
+        model_dict = self.dict()
+        for k in model_dict:
+            new_key = self.get_env_var_name(attr_name=k)
+            if isinstance((old_val := model_dict[k]), SecretStr):
+                new_val = old_val.get_secret_value()
+            else:
+                new_val = old_val
+            result[new_key] = new_val
+        return result
+
+    def dump(self, target: Union[str, Path]) -> None:
+        target_path = Path(target)
+        ser_dict = self._get_serializable_dict()
+        with open(target_path, "w") as target_file:
+            for k in ser_dict:
+                if (val := ser_dict[k]) is not None:  # only dump existing values
+                    target_file.write(f'{k}="{val}"\n')
+
+
+class ProfileSettings(DumpableSettings):
+    host: str
+    username: str
+    api_key: SecretStr
+    verify_ssl: bool = True
+
+    class Config:
+        env_prefix = "DEEPSEARCH_"
+
+    @classmethod
+    def from_cli_prompt(cls) -> ProfileSettings:
+        return cls(
+            host=input("Host: "),
+            username=input("Username: "),
+            api_key=getpass("API key: "),
+            verify_ssl=input("SSL verification [y/n]: "),
+        )
+
+
+class MainSettings(DumpableSettings):
+
+    profile: Optional[str] = None  # None only when profiles not yet iniitialized
+    show_cli_stack_traces: bool = False
+
+    class Config:
+        env_prefix = "DEEPSEARCH_"