prefect.cli.cloud

Command line interface for interacting with Prefect Cloud

CloudUnauthorizedError

Raised when the CloudClient receives a 401 or 403 from the Cloud API.

Source code in prefect/cli/cloud.py

class CloudUnauthorizedError(PrefectException):
    """
    Raised when the CloudClient receives a 401 or 403 from the Cloud API.
    """

build_table

Generate a table of workspaces. The select_idx of workspaces will be highlighted.

Parameters:

Name	Description	Default
selected_idx	currently selected index int	required
workspaces	Iterable of strings Iterable[str]	required

Returns:

Type	Description
Table	rich.table.Table

Source code in prefect/cli/cloud.py

def build_table(selected_idx: int, workspaces: Iterable[str]) -> Table:
    """
    Generate a table of workspaces. The `select_idx` of workspaces will be highlighted.

    Args:
        selected_idx: currently selected index
        workspaces: Iterable of strings

    Returns:
        rich.table.Table
    """

    table = Table()
    table.add_column(
        "[#024dfd]Select a Workspace:", justify="right", style="#8ea0ae", no_wrap=True
    )

    for i, workspace in enumerate(sorted(workspaces)):
        if i == selected_idx:
            table.add_row("[#024dfd on #FFFFFF]> " + workspace)
        else:
            table.add_row("  " + workspace)
    return table

login async

Log in to Prefect Cloud. Sets PREFECT_API_URL and PREFECT_API_KEY for profile. If those values are already set they will be overwritten.

Source code in prefect/cli/cloud.py

@cloud_app.command()
async def login(
    key: str = typer.Option(
        ..., "--key", "-k", help="API Key to authenticate with Prefect", prompt=True
    ),
    workspace_handle: str = typer.Option(
        None,
        "--workspace",
        "-w",
        help="Full handle of workspace, in format '<account_handle>/<workspace_handle>'",
    ),
):
    """
    Log in to Prefect Cloud.
    Sets PREFECT_API_URL and PREFECT_API_KEY for profile.
    If those values are already set they will be overwritten.
    """

    async with get_cloud_client(api_key=key) as client:
        try:
            workspaces = await client.read_workspaces()
        except CloudUnauthorizedError:
            exit_with_error(
                "Unable to authenticate. Please ensure your credentials are correct."
            )

    workspaces = {
        f"{workspace['account_handle']}/{workspace['workspace_handle']}": workspace
        for workspace in workspaces
    }

    if not workspace_handle:
        workspace_handle = select_workspace(workspaces.keys())

    if workspace_handle not in workspaces:
        exit_with_error(
            f"Workspace {workspace_handle!r} not found. "
            "Leave `--workspace` blank to select a workspace."
        )

    profile = update_current_profile(
        {
            PREFECT_API_URL: build_url_from_workspace(workspaces[workspace_handle]),
            PREFECT_API_KEY: key,
        }
    )

    exit_with_success(
        "Successfully logged in and set workspace to "
        f"{workspace_handle!r} in profile {profile.name!r}."
    )

logout async

Log out of Prefect Cloud. Removes PREFECT_API_URL and PREFECT_API_KEY from profile.

Source code in prefect/cli/cloud.py

@cloud_app.command()
async def logout():
    """
    Log out of Prefect Cloud.
    Removes PREFECT_API_URL and PREFECT_API_KEY from profile.
    """
    confirm_logged_in()
    profile = update_current_profile({PREFECT_API_URL: None, PREFECT_API_KEY: None})
    exit_with_success(f"Successfully logged out in profile {profile.name!r}.")

select_workspace

Given a list of workspaces, display them to user in a Table and allow them to select one.

Parameters:

Name	Description	Default
workspaces	List of workspaces to choose from Iterable[str]	required

Returns:

Type	Description
str	the selected workspace

Source code in prefect/cli/cloud.py

def select_workspace(workspaces: Iterable[str]) -> str:
    """
    Given a list of workspaces, display them to user in a Table
    and allow them to select one.

    Args:
        workspaces: List of workspaces to choose from

    Returns:
        str: the selected workspace
    """

    workspaces = sorted(workspaces)
    current_idx = 0
    selected_workspace = None

    with Live(
        build_table(current_idx, workspaces), auto_refresh=False, console=app.console
    ) as live:
        while selected_workspace is None:
            key = readchar.readkey()

            if key == readchar.key.UP:
                current_idx = current_idx - 1
                # wrap to bottom if at the top
                if current_idx < 0:
                    current_idx = len(workspaces) - 1
            elif key == readchar.key.DOWN:
                current_idx = current_idx + 1
                # wrap to top if at the bottom
                if current_idx >= len(workspaces):
                    current_idx = 0
            elif key == readchar.key.CTRL_C:
                # gracefully exit with no message
                exit_with_error("")
            elif key == readchar.key.ENTER:
                selected_workspace = workspaces[current_idx]

            live.update(build_table(current_idx, workspaces), refresh=True)

        return selected_workspace

set async

Set current workspace.

Source code in prefect/cli/cloud.py

@workspace_app.command()
async def set(
    workspace_handle: str = typer.Option(
        None,
        "--workspace",
        "-w",
        help="Full handle of workspace, in format '<account_handle>/<workspace_handle>'",
    ),
):
    """Set current workspace."""
    confirm_logged_in()

    async with get_cloud_client() as client:
        try:
            workspaces = await client.read_workspaces()
        except CloudUnauthorizedError:
            exit_with_error(
                "Unable to authenticate. Please ensure your credentials are correct."
            )
    workspaces = {
        f"{workspace['account_handle']}/{workspace['workspace_handle']}": workspace
        for workspace in workspaces
    }

    if not workspace_handle:
        workspace_handle = select_workspace(workspaces)

    if workspace_handle not in workspaces:
        exit_with_error(
            f"Workspace {workspace_handle!r} not found. "
            "Leave `--workspace` blank to select a workspace."
        )

    profile = update_current_profile(
        {PREFECT_API_URL: build_url_from_workspace(workspaces[workspace_handle])}
    )

    exit_with_success(
        f"Successfully set workspace to {workspace_handle!r} in profile {profile.name!r}."
    )