diff --git a/workflow/cli/main.py b/workflow/cli/main.py index ca126bb..9791721 100755 --- a/workflow/cli/main.py +++ b/workflow/cli/main.py @@ -1,6 +1,7 @@ """Workflow command line interface.""" import click +from rich.console import Console # from workflow.cli.buckets import buckets from workflow.cli.pipelines import pipelines @@ -8,11 +9,17 @@ from workflow.cli.run import run from workflow.cli.schedules import schedules from workflow.cli.workspace import workspace +from workflow.utils.read import get_active_workspace + +console = Console() @click.group() def cli(): """Workflow Command Line Interface.""" + # ? Get workspace + message = get_active_workspace() + console.print(message) pass diff --git a/workflow/utils/read.py b/workflow/utils/read.py index 81b5403..4679f52 100644 --- a/workflow/utils/read.py +++ b/workflow/utils/read.py @@ -6,8 +6,10 @@ from requests import get from requests.exceptions import RequestException +from rich.text import Text from yaml import safe_load +from workflow.cli.workspace import localspaces, localstems, modulespaces, modulestems from workflow.utils.logger import get_logger logger = get_logger("workflow.utils.read") @@ -83,6 +85,33 @@ def filename(source: str) -> Any: raise error +def get_active_workspace() -> Text: + """Returns a Text with info about the active workspace. + + Returns + ------- + Text + Text instance with info. + """ + _workspace = "active" + text = Text() + if _workspace in localstems: + for possibility in localspaces.glob(f"{_workspace}.y*ml"): + config = workspace(possibility) + text.append("Currently using ", style="green") + text.append(f"{config['workspace']} ", style="blue") + text.append("workspace.", style="green") + elif _workspace in modulestems: + for possibility in modulespaces.glob(f"{_workspace}.y*ml"): + config = workspace(possibility) + text.append("Currently using ", style="green") + text.append(f"{config['workspace']} ", style="blue") + text.append("workspace.", style="green") + else: + text.append("There is not active workspace", style="red") + return text + + def is_valid_url(url: str) -> bool: """Return True if the URL is valid.