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}."
)