Browse Source

Use new token enable endpoint, pydantic

master
Teran McKinney 4 weeks ago
parent
commit
75fecd72ba
  1. 7
      mypy.ini
  2. 1
      setup.py
  3. 181
      src/sporestack/api.py
  4. 25
      src/sporestack/api_client.py
  5. 39
      src/sporestack/client.py
  6. 22
      src/sporestack/models.py

7
mypy.ini

@ -1,5 +1,12 @@
[mypy]
exclude = build
plugins = pydantic.mypy
[pydantic-mypy]
init_forbid_extra = True
init_typed = True
warn_required_dynamic_aliases = True
warn_untyped_fields = True
[mypy-pytest.*]
ignore_missing_imports = True

1
setup.py

@ -47,6 +47,7 @@ setup(
"segno",
"requests[socks]>=2.22.0",
"typer",
"pydantic",
],
entry_points={"console_scripts": ["sporestack = sporestack.client:cli"]},
zip_safe=False, # This is so py.typed gets included.

181
src/sporestack/api.py

@ -0,0 +1,181 @@
"""
SporeStack API request/response models
"""
from typing import List, Optional
from pydantic import BaseModel
from .models import NetworkInterface, Payment
class deprecated_settlement_enable(object):
url = "/settlement/enable"
method = "POST"
class Request(BaseModel):
settlement_token: str
currency: str
cents: int
class Response(BaseModel):
payment_uri: Optional[str]
txid: Optional[str]
paid: bool
usd: str
class deprecated_settlement_add(object):
url = "/settlement/add"
method = "POST"
class Request(BaseModel):
settlement_token: str
currency: str
cents: int
class Response(BaseModel):
payment_uri: Optional[str]
txid: Optional[str]
paid: bool
usd: str
class deprecated_settlement_balance(object):
url = "/settlement/balance"
method = "GET"
class Response(BaseModel):
cents: int
usd: str
class token_enable(object):
url = "/token/{token}/enable"
method = "POST"
class Request(BaseModel):
currency: str
dollars: int
class Response(BaseModel):
token: str
payment: Payment
class token_add(object):
url = "/token/{token}/add"
method = "POST"
class Request(BaseModel):
currency: str
dollars: int
class Response(BaseModel):
token: str
payment: Payment
class token_balance(object):
url = "/token/{token}/balance"
method = "GET"
class Response(BaseModel):
token: str
cents: int
usd: str
class server_delete(object):
url = "/v2/delete"
method = "POST"
class Request(BaseModel):
machine_id: str
class server_launch(object):
url = "/v2/launch"
method = "POST"
class Request(BaseModel):
machine_id: str
days: int
currency: str
flavor: str
ssh_key: str
operating_system: str
region: Optional[str]
organization: Optional[str]
settlement_token: Optional[str]
affiliate_amount: Optional[int]
affiliate_token: Optional[str]
class Response(BaseModel):
created_at: Optional[int]
payment: Payment
expiration: Optional[int]
machine_id: str
network_interfaces: List[NetworkInterface]
region: str
latest_api_version: int
created: bool
paid: bool
warning: Optional[str]
txid: Optional[str]
operating_system: str
flavor: str
class server_start(object):
url = "/v2/start"
method = "POST"
class Request(BaseModel):
machine_id: str
class server_stop(object):
url = "/v2/stop"
method = "POST"
class Request(BaseModel):
machine_id: str
class server_topup(object):
url = "/v2/topup"
method = "POST"
class Request(BaseModel):
machine_id: str
days: int
currency: str
settlement_token: Optional[str]
affiliate_amount: Optional[int]
affiliate_token: Optional[str]
class Response(BaseModel):
machine_id: str
payment: Payment
paid: bool
warning: Optional[str]
expiration: int
txid: Optional[str]
latest_api_version: int
class server_info(object):
url = "/v2/info"
method = "GET"
class Response(BaseModel):
created_at: int
expiration: int
running: bool
machine_id: str
network_interfaces: List[NetworkInterface]
region: str

25
src/sporestack/api_client.py

@ -4,7 +4,7 @@ from typing import Any, Dict, Optional
import requests
from . import validate
from . import api, validate
log = logging.getLogger(__name__)
@ -222,23 +222,18 @@ def info(machine_id: str, api_endpoint: str = API_ENDPOINT) -> Any:
return api_request(url, get_params=get_params)
def settlement_token_enable(
settlement_token: str,
cents: int,
def token_enable(
token: str,
dollars: int,
currency: str,
api_endpoint: str = API_ENDPOINT,
retry: bool = False,
) -> Any:
validate.settlement_token(settlement_token)
validate.cents(cents)
json_params = {
"settlement_token": settlement_token,
"cents": cents,
"currency": currency,
}
url = api_endpoint + "/settlement/enable"
return api_request(url=url, json_params=json_params, retry=retry)
) -> api.token_enable.Response:
request = api.token_enable.Request(dollars=dollars, currency=currency)
url = api_endpoint + api.token_enable.url.format(token=token)
response = api_request(url=url, json_params=request.dict(), retry=retry)
assert response.token == token
return api.token_enable.Response.parse_obj(response)
def settlement_token_add(

39
src/sporestack/client.py

@ -431,35 +431,38 @@ def settlement_token_enable(
"""
Enables a new settlement token.
Cents is starting balance.
Dollars is starting balance.
"""
cents = dollars * 100
def enable_token() -> Any:
return api_client.settlement_token_enable(
settlement_token=token,
cents=cents,
currency=currency,
api_endpoint=api_endpoint,
retry=True,
)
response = api_client.token_enable(
token=token,
dollars=dollars,
currency=currency,
api_endpoint=api_endpoint,
retry=True,
)
enable_dict = enable_token()
uri = enable_dict["payment_uri"]
usd = enable_dict["usd"]
uri = response.payment.uri
assert uri is not None
usd = response.payment.usd
make_payment(currency=currency, uri=uri, usd=usd)
tries = 360
tries = 360 * 2
while tries > 0:
print(WAITING_PAYMENT_TO_PROCESS, file=sys.stderr)
tries = tries - 1
# FIXME: Wait one hour in a smarter way.
# FIXME: Wait two hours in a smarter way.
# Waiting for payment to set in.
time.sleep(10)
enable_dict = enable_token()
if enable_dict["paid"] is True:
response = api_client.token_enable(
token=token,
dollars=dollars,
currency=currency,
api_endpoint=api_endpoint,
retry=True,
)
if response.payment.paid is True:
break
print(f"{token} has been enabled. Save it and don't lose it!")

22
src/sporestack/models.py

@ -0,0 +1,22 @@
"""
SporeStack API supplemental models
"""
from typing import Optional
from pydantic import BaseModel
class NetworkInterface(BaseModel):
ipv4: str
ipv6: str
class Payment(BaseModel):
txid: Optional[str]
uri: Optional[str]
usd: str
paid: bool
Loading…
Cancel
Save