@ -5,14 +5,21 @@ SporeStack CLI: `sporestack`
import json
import logging
import os
import sys
import time
from pathlib import Path
from typing import TYPE_CHECKING , Any , Dict , Optional
import typer
if sys . version_info > = ( 3 , 9 ) : # pragma: nocover
from typing import Annotated
else : # pragma: nocover
from typing_extensions import Annotated
if TYPE_CHECKING :
from . import api
from . models import Invoice
HELP = """
@ -71,15 +78,21 @@ def get_api_endpoint() -> str:
return api_endpoint
def make_payment ( currency: str , uri : str , usd : str ) - > None :
def make_payment ( invoice: " Invoice " ) - > None :
import segno
premessage = """ Payment URI: {}
Pay * exactly * the specified amount . No more , no less . Pay within
one hour at the very most .
from . _cli_utils import cents_to_usd
uri = invoice . payment_uri
usd = cents_to_usd ( invoice . amount )
expires = epoch_to_human ( invoice . expires )
message = f """ Invoice: { invoice . id }
Invoice expires : { expires } ( payment must be confirmed by this time )
Payment URI : { uri }
Pay * exactly * the specified amount . No more , no less .
Resize your terminal and try again if QR code above is not readable .
Press ctrl + c to abort . """
message = premessage . format ( uri )
qr = segno . make ( uri )
# This typer.echos.
qr . terminal ( )
@ -90,11 +103,12 @@ Press ctrl+c to abort."""
@server_cli.command ( )
def launch (
days : Annotated [
int ,
typer . Option ( min = 1 , max = 90 , help = " Number of days the server should run for. " ) ,
] ,
operating_system : Annotated [ str , typer . Option ( help = " Example: debian-11 " ) ] ,
hostname : str = " " ,
days : int = typer . Option (
. . . , min = 1 , max = 90 , help = " Number of days the server should run for. "
) ,
operating_system : str = typer . Option ( . . . , help = " Example: debian-11 " ) ,
ssh_key_file : Path = DEFAULT_SSH_KEY_FILE ,
flavor : str = DEFAULT_FLAVOR ,
token : str = DEFAULT_TOKEN ,
@ -301,7 +315,7 @@ def print_machine_info(info: "api.ServerInfo.Response") -> None:
typer . echo ( f " Region: { info . region } " )
typer . echo ( f " Flavor: { info . flavor . slug } " )
typer . echo ( f " Expiration: { epoch_to_human ( info . expiration ) } " )
typer . echo ( f " Token : { info . token } " )
typer . echo ( f " Token (keep this secret!) : { info . token } " )
if info . deleted_at != 0 or info . deleted :
typer . echo ( " Server was deleted! " )
if info . deleted_at != 0 :
@ -321,20 +335,35 @@ def print_machine_info(info: "api.ServerInfo.Response") -> None:
@server_cli.command ( name = " list " )
def server_list (
token : str = DEFAULT_TOKEN ,
local : bool = typer . Option (
True , help = " List older servers not associated to token. "
) ,
show_forgotten : bool = typer . Option (
False , help = " Show deleted and forgotten servers. "
) ,
local : Annotated [
bool , typer . Option ( help = " List older servers not associated to token. " )
] = True ,
show_forgotten : Annotated [
bool , typer . Option ( help = " Show deleted and forgotten servers. " )
] = False ,
) - > None :
""" List all locally known servers and all servers under the given token. """
""" Lists a token ' s servers. """
_token = load_token ( token )
from rich . console import Console
from rich . table import Table
from . api_client import APIClient
from . exceptions import SporeStackUserError
api_client = APIClient ( api_endpoint = get_api_endpoint ( ) )
console = Console ( width = None if sys . stdout . isatty ( ) else 10 * * 9 )
_token = load_token ( token )
table = Table (
title = f " Servers for { token } ( { _token } ) " ,
show_header = True ,
header_style = " bold magenta " ,
caption = (
" For more details on a server, run "
" `sporestack server info --machine-id (machine id)` "
) ,
)
api_client = APIClient ( api_endpoint = get_api_endpoint ( ) )
server_infos = api_client . servers_launched_from_token ( token = _token ) . servers
machine_id_hostnames = { }
@ -348,6 +377,13 @@ def server_list(
printed_machine_ids = [ ]
table . add_column ( " Machine ID [bold](Secret!)[/bold] " , style = " dim " )
table . add_column ( " Hostname " )
table . add_column ( " IPv4 " )
table . add_column ( " IPv6 " )
table . add_column ( " Expires At " )
table . add_column ( " Autorenew " )
for info in server_infos :
if not show_forgotten and info . forgotten_at is not None :
continue
@ -360,10 +396,23 @@ def server_list(
hostname = machine_id_hostnames [ info . machine_id ]
info . hostname = hostname
print_machine_info ( info )
expiration = epoch_to_human ( info . expiration )
if info . deleted_at :
expiration = f " [bold]Deleted[/bold] at { epoch_to_human ( info . deleted_at ) } "
table . add_row (
info . machine_id ,
info . hostname ,
info . ipv4 ,
info . ipv6 ,
expiration ,
str ( info . autorenew ) ,
)
printed_machine_ids . append ( info . machine_id )
console . print ( table )
if local :
for hostname_json in os . listdir ( directory ) :
hostname = hostname_json . split ( " . " ) [ 0 ]
@ -383,7 +432,7 @@ def server_list(
except SporeStackUserError as e :
expiration = saved_vm_info [ " expiration " ]
human_expiration = time . strftime (
" % Y- % m- %d % H: % M: % S % z " , time . localtime ( )
" % Y- % m- %d % H: % M: % S % z " , time . localtime ( saved_vm_info[ " expiration" ] )
)
msg = hostname
msg + = f " expired ( { expiration } { human_expiration } ): "
@ -562,7 +611,7 @@ def flavors() -> None:
from . _cli_utils import cents_to_usd , gb_string , mb_string , tb_string
from . api_client import APIClient
console = Console ( )
console = Console ( width = None if sys . stdout . isatty ( ) else 10 * * 9 )
table = Table ( show_header = True , header_style = " bold magenta " )
table . add_column ( " Flavor Slug (--flavor) " )
@ -664,15 +713,17 @@ def save_token(token: str, key: str) -> None:
@token_cli.command ( name = " create " )
def token_create (
token: str = typer . Argument ( DEFAULT_TOKEN ) ,
dollars: int = typer . Option ( . . . ) ,
currency: str = typer . Option ( . . . ) ,
dollars: Annotated [ int , typer . Option ( ) ] ,
currency: Annotated [ str , typer . Option ( ) ] ,
token: Annotated [ str , typer . Argument ( ) ] = DEFAULT_TOKEN ,
) - > None :
"""
Enables a new token .
Dollars is starting balance .
"""
from httpx import HTTPError
from . import utils
_token = utils . random_token ( )
@ -694,11 +745,7 @@ def token_create(
currency = currency ,
)
uri = response . payment . uri
assert uri is not None
usd = response . payment . usd
make_payment ( currency = currency , uri = uri , usd = usd )
make_payment ( response . invoice )
tries = 360 * 2
while tries > 0 :
@ -713,10 +760,10 @@ def token_create(
dollars = dollars ,
currency = currency ,
)
except SporeStackServerError :
except ( SporeStackServerError , HTTPError ) :
typer . echo ( " Received 500 HTTP status, will try again. " , err = True )
continue
if response . payment. paid is True :
if response . invoice. paid :
typer . echo ( f " { token } has been enabled with $ { dollars } . " )
typer . echo ( f " { token } ' s key is { _token } . " )
typer . echo ( " Save it, don ' t share it, and don ' t lose it! " )
@ -743,6 +790,8 @@ def token_topup(
""" Adds balance to an existing token. """
token = load_token ( token )
from httpx import HTTPError
from . api_client import APIClient
from . exceptions import SporeStackServerError
@ -754,11 +803,7 @@ def token_topup(
currency = currency ,
)
uri = response . payment . uri
assert uri is not None
usd = response . payment . usd
make_payment ( currency = currency , uri = uri , usd = usd )
make_payment ( response . invoice )
tries = 360 * 2
while tries > 0 :
@ -771,12 +816,12 @@ def token_topup(
dollars = dollars ,
currency = currency ,
)
except SporeStackServerError :
except ( SporeStackServerError , HTTPError ) :
typer . echo ( " Received 500 HTTP status, will try again. " , err = True )
continue
# Waiting for payment to set in.
time . sleep ( 10 )
if response . payment. paid is True :
if response . invoice. paid :
typer . echo ( f " Added { dollars } dollars to { token } " )
return
raise ValueError ( f " { token } did not get enabled in time. " )
@ -795,7 +840,7 @@ def balance(token: str = typer.Argument(DEFAULT_TOKEN)) -> None:
@token_cli.command ( name = " info " )
def token_info ( token : str = typer . Argument ( DEFAULT_TOKEN ) ) - > None :
def token_info ( token : Annotated [ str , typer . Argument ( ) ] = DEFAULT_TOKEN ) - > None :
"""
Show information about a token , including balance .
@ -828,7 +873,7 @@ def token_info(token: str = typer.Argument(DEFAULT_TOKEN)) -> None:
@token_cli.command ( )
def servers ( token : str = typer . Argument ( DEFAULT_TOKEN ) ) - > None :
def servers ( token : Annotated [ str , typer . Argument ( ) ] = DEFAULT_TOKEN ) - > None :
""" Returns server info for servers launched by a given token. """
_token = load_token ( token )
@ -842,13 +887,75 @@ def servers(token: str = typer.Argument(DEFAULT_TOKEN)) -> None:
@token_cli.command ( name = " list " )
def token_list ( ) - > None :
""" List tokens. """
from rich . console import Console
from rich . table import Table
console = Console ( width = None if sys . stdout . isatty ( ) else 10 * * 9 )
token_dir = token_path ( )
typer . echo ( f " SporeStack tokens present in { token_dir } : " , err = True )
typer . echo ( " (Name): (Key) " , err = True )
table = Table (
show_header = True ,
header_style = " bold magenta " ,
caption = f " These tokens are stored in { token_dir } " ,
)
table . add_column ( " Name " )
table . add_column ( " Token (this is a globally unique [bold]secret[/bold]) " )
for token_file in token_dir . glob ( " *.json " ) :
token = token_file . stem
key = load_token ( token )
typer . echo ( f " { token } : { key } " )
table . add_row ( token , key )
console . print ( table )
@token_cli.command ( name = " invoices " )
def token_invoices ( token : Annotated [ str , typer . Argument ( ) ] = DEFAULT_TOKEN ) - > None :
""" List invoices. """
_token = load_token ( token )
from rich . console import Console
from rich . table import Table
from . _cli_utils import cents_to_usd
from . api_client import APIClient
from . client import Client
api_client = APIClient ( api_endpoint = get_api_endpoint ( ) )
client = Client ( api_client = api_client , client_token = _token )
console = Console ( width = None if sys . stdout . isatty ( ) else 10 * * 9 )
table = Table (
title = f " Invoices for { token } ( { _token } ) " ,
show_header = True ,
header_style = " bold magenta " ,
)
table . add_column ( " ID " )
table . add_column ( " Amount " )
table . add_column ( " Created At " )
table . add_column ( " Paid At " )
table . add_column ( " URI " )
table . add_column ( " TXID " )
for invoice in client . token . invoices ( ) :
if invoice . paid :
paid = epoch_to_human ( invoice . paid )
else :
if invoice . expired :
paid = " [bold]Expired[/bold] "
else :
paid = f " Unpaid. Expires: { epoch_to_human ( invoice . expires ) } "
table . add_row (
str ( invoice . id ) ,
cents_to_usd ( invoice . amount ) ,
epoch_to_human ( invoice . created ) ,
paid ,
invoice . payment_uri ,
invoice . txid ,
)
console . print ( table )
@token_cli.command ( )
SporeStack
3 weeks ago