diff --git a/sdks/python/setup.py b/sdks/python/setup.py index f6c70cbe2d..73d513259f 100644 --- a/sdks/python/setup.py +++ b/sdks/python/setup.py @@ -47,7 +47,6 @@ "pytest", "rich", "tqdm", - "questionary", "uuid7<1.0.0", ], entry_points={ diff --git a/sdks/python/src/opik/cli.py b/sdks/python/src/opik/cli.py index 071348b8ef..37b68b8d4d 100644 --- a/sdks/python/src/opik/cli.py +++ b/sdks/python/src/opik/cli.py @@ -1,11 +1,13 @@ """CLI tool for Opik.""" +import logging from importlib import metadata import click -import questionary - from opik.configurator import configure as opik_configure +from opik.configurator import interactive_helpers + +LOGGER = logging.getLogger(__name__) __version__: str = "0.0.0+dev" if __package__: @@ -35,30 +37,30 @@ def configure(use_local: bool) -> None: if use_local: opik_configure.configure(use_local=True, force=True) else: - deployment_type_choice = questionary.select( - "Which Opik deployment do you want to log your traces to?", - choices=["Opik Cloud", "Self-hosted Comet platform", "Local deployment"], - ).unsafe_ask() + deployment_type_choice = interactive_helpers.ask_user_for_deployment_type() - if deployment_type_choice == "Opik Cloud": + if deployment_type_choice == interactive_helpers.DeploymentType.CLOUD: configurator = opik_configure.OpikConfigurator( url=opik_configure.OPIK_BASE_URL_CLOUD, use_local=False, force=True, self_hosted_comet=False, ) - elif deployment_type_choice == "Self-hosted Comet platform": + elif deployment_type_choice == interactive_helpers.DeploymentType.SELF_HOSTED: configurator = opik_configure.OpikConfigurator( use_local=False, force=True, self_hosted_comet=True, ) - else: + elif deployment_type_choice == interactive_helpers.DeploymentType.LOCAL: configurator = opik_configure.OpikConfigurator( use_local=True, force=True, self_hosted_comet=False, ) + else: + LOGGER.error("Unknown deployment type was selected. Exiting.") + exit(1) configurator.configure() diff --git a/sdks/python/src/opik/configurator/interactive_helpers.py b/sdks/python/src/opik/configurator/interactive_helpers.py index 6b7d5409e5..6ffbbfd1bf 100644 --- a/sdks/python/src/opik/configurator/interactive_helpers.py +++ b/sdks/python/src/opik/configurator/interactive_helpers.py @@ -1,3 +1,4 @@ +import enum import logging import sys @@ -95,3 +96,56 @@ def ask_user_for_approval(message: str) -> bool: if users_choice in ("N", "NO"): return False LOGGER.error("Wrong choice. Please try again.") + + +class DeploymentType(enum.Enum): + CLOUD = (1, "Opik Cloud (default)") + SELF_HOSTED = (2, "Self-hosted Comet platform") + LOCAL = (3, "Local deployment") + + @classmethod + def find_by_value(cls, value: int) -> "DeploymentType": + """ + Find the DeploymentType by its integer value. + + :param value: The integer value of the DeploymentType. + :return: The corresponding DeploymentType. + """ + for v in cls: + if v.value[0] == value: + return v + raise ValueError(f"No DeploymentType with value '{value}'") + + +def ask_user_for_deployment_type() -> DeploymentType: + """ + Asks the user to select a deployment type from the available Opik deployment options. + Prompts the user until a valid selection is made. + + Returns: + DeploymentType: The user's selected deployment type. + """ + msg = ["Which Opik deployment do you want to log your traces to?"] + + for deployment in DeploymentType: + msg.append(f"{deployment.value[0]} - {deployment.value[1]}") + + msg.append("\n> ") + + message_string = "\n".join(msg) + + while True: + choice_str = input(message_string).strip() + + if choice_str not in ("1", "2", "3", ""): + LOGGER.error("Wrong choice. Please try again.\n") + continue + + if choice_str == "": + choice_index = 1 + else: + choice_index = int(choice_str) + + choice = DeploymentType.find_by_value(choice_index) + + return choice