332 lines
12 KiB
Python
332 lines
12 KiB
Python
import logging
|
|
import os
|
|
from dataclasses import dataclass
|
|
from typing import List, Optional, Union
|
|
|
|
import httpx
|
|
from pydantic import parse_obj_as
|
|
|
|
from . import __version__, api, exceptions
|
|
from .models import Currency, Invoice, ServerUpdateRequest, TokenInfo
|
|
|
|
log = logging.getLogger(__name__)
|
|
|
|
LATEST_API_VERSION = 2
|
|
"""This is probably not used anymore."""
|
|
|
|
CLEARNET_ENDPOINT = "https://api.sporestack.com"
|
|
TOR_ENDPOINT = (
|
|
"http://api.spore64i5sofqlfz5gq2ju4msgzojjwifls7rok2cti624zyq3fcelad.onion"
|
|
)
|
|
|
|
API_ENDPOINT = CLEARNET_ENDPOINT
|
|
|
|
TIMEOUT = httpx.Timeout(60.0)
|
|
|
|
HEADERS = {"User-Agent": f"sporestack-python/{__version__}"}
|
|
|
|
|
|
def _get_tor_proxy() -> str:
|
|
"""
|
|
This makes testing easier.
|
|
"""
|
|
return os.getenv("TOR_PROXY", "socks5://127.0.0.1:9050")
|
|
|
|
|
|
# For requests module
|
|
TOR_PROXY_REQUESTS = {"http": _get_tor_proxy(), "https": _get_tor_proxy()}
|
|
|
|
|
|
def _is_onion_url(url: str) -> bool:
|
|
"""
|
|
returns True/False depending on if a URL looks like a Tor hidden service
|
|
(.onion) or not.
|
|
This is designed to false as non-onion just to be on the safe-ish side,
|
|
depending on your view point. It requires URLs like: http://domain.tld/,
|
|
not http://domain.tld or domain.tld/.
|
|
|
|
This can be optimized a lot.
|
|
"""
|
|
try:
|
|
url_parts = url.split("/")
|
|
domain = url_parts[2]
|
|
tld = domain.split(".")[-1]
|
|
if tld == "onion":
|
|
return True
|
|
except Exception:
|
|
pass
|
|
return False
|
|
|
|
|
|
def _get_response_error_text(response: httpx.Response) -> str:
|
|
"""Get a response's error text. Assumes the response is actually an error."""
|
|
if (
|
|
"content-type" in response.headers
|
|
and response.headers["content-type"] == "application/json"
|
|
):
|
|
error = response.json()
|
|
if "detail" in error:
|
|
if isinstance(error["detail"], str):
|
|
return error["detail"]
|
|
else:
|
|
return str(error["detail"])
|
|
|
|
return response.text
|
|
|
|
|
|
def _handle_response(response: httpx.Response) -> None:
|
|
status_code_first_digit = response.status_code // 100
|
|
if status_code_first_digit == 2:
|
|
return
|
|
|
|
error_response_text = _get_response_error_text(response)
|
|
if response.status_code == 429:
|
|
raise exceptions.SporeStackTooManyRequestsError(error_response_text)
|
|
elif status_code_first_digit == 4:
|
|
raise exceptions.SporeStackUserError(error_response_text)
|
|
elif status_code_first_digit == 5:
|
|
# User should probably retry.
|
|
raise exceptions.SporeStackServerError(error_response_text)
|
|
else:
|
|
# This would be weird.
|
|
raise exceptions.SporeStackServerError(error_response_text)
|
|
|
|
|
|
@dataclass
|
|
class APIClient:
|
|
api_endpoint: str = API_ENDPOINT
|
|
|
|
def __post_init__(self) -> None:
|
|
headers = httpx.Headers(HEADERS)
|
|
proxy = None
|
|
if _is_onion_url(self.api_endpoint):
|
|
proxy = _get_tor_proxy()
|
|
self._httpx_client = httpx.Client(
|
|
headers=headers, proxies=proxy, timeout=TIMEOUT
|
|
)
|
|
|
|
def server_launch(
|
|
self,
|
|
machine_id: str,
|
|
days: int,
|
|
flavor: str,
|
|
operating_system: str,
|
|
ssh_key: str,
|
|
token: str,
|
|
region: Optional[str] = None,
|
|
hostname: str = "",
|
|
autorenew: bool = False,
|
|
) -> None:
|
|
"""Launch a server."""
|
|
request = api.ServerLaunch.Request(
|
|
days=days,
|
|
token=token,
|
|
flavor=flavor,
|
|
region=region,
|
|
operating_system=operating_system,
|
|
ssh_key=ssh_key,
|
|
hostname=hostname,
|
|
autorenew=autorenew,
|
|
)
|
|
url = self.api_endpoint + api.ServerLaunch.url.format(machine_id=machine_id)
|
|
response = self._httpx_client.post(url=url, json=request.dict())
|
|
_handle_response(response)
|
|
|
|
def server_topup(
|
|
self,
|
|
machine_id: str,
|
|
days: int,
|
|
token: Union[str, None] = None,
|
|
) -> None:
|
|
"""Topup a server."""
|
|
request = api.ServerTopup.Request(days=days, token=token)
|
|
url = self.api_endpoint + api.ServerTopup.url.format(machine_id=machine_id)
|
|
response = self._httpx_client.post(url=url, json=request.dict())
|
|
_handle_response(response)
|
|
|
|
def server_quote(self, days: int, flavor: str) -> api.ServerQuote.Response:
|
|
"""Get a quote for how much a server will cost."""
|
|
|
|
url = self.api_endpoint + api.ServerQuote.url
|
|
response = self._httpx_client.get(
|
|
url,
|
|
params={"days": days, "flavor": flavor},
|
|
)
|
|
_handle_response(response)
|
|
return api.ServerQuote.Response.parse_obj(response.json())
|
|
|
|
def autorenew_enable(self, machine_id: str) -> None:
|
|
"""Enable autorenew on a server."""
|
|
url = self.api_endpoint + api.ServerEnableAutorenew.url.format(
|
|
machine_id=machine_id
|
|
)
|
|
response = self._httpx_client.post(url)
|
|
_handle_response(response)
|
|
|
|
def autorenew_disable(self, machine_id: str) -> None:
|
|
"""Disable autorenew on a server."""
|
|
url = self.api_endpoint + api.ServerDisableAutorenew.url.format(
|
|
machine_id=machine_id
|
|
)
|
|
response = self._httpx_client.post(url)
|
|
_handle_response(response)
|
|
|
|
def server_start(self, machine_id: str) -> None:
|
|
"""Power on a server."""
|
|
url = self.api_endpoint + api.ServerStart.url.format(machine_id=machine_id)
|
|
response = self._httpx_client.post(url)
|
|
_handle_response(response)
|
|
|
|
def server_stop(self, machine_id: str) -> None:
|
|
"""Power off a server."""
|
|
url = self.api_endpoint + api.ServerStop.url.format(machine_id=machine_id)
|
|
response = self._httpx_client.post(url)
|
|
_handle_response(response)
|
|
|
|
def server_delete(self, machine_id: str) -> None:
|
|
"""Delete a server."""
|
|
url = self.api_endpoint + api.ServerDelete.url.format(machine_id=machine_id)
|
|
response = self._httpx_client.delete(url)
|
|
_handle_response(response)
|
|
|
|
def server_forget(self, machine_id: str) -> None:
|
|
"""Forget about a deleted server to hide it from view."""
|
|
url = self.api_endpoint + api.ServerForget.url.format(machine_id=machine_id)
|
|
response = self._httpx_client.post(url)
|
|
_handle_response(response)
|
|
|
|
def server_rebuild(self, machine_id: str) -> None:
|
|
"""
|
|
Rebuilds the server with the operating system and SSH key set at launch time.
|
|
|
|
Deletes all of the data on the server!
|
|
"""
|
|
url = self.api_endpoint + api.ServerRebuild.url.format(machine_id=machine_id)
|
|
response = self._httpx_client.post(url)
|
|
_handle_response(response)
|
|
|
|
def server_info(self, machine_id: str) -> api.ServerInfo.Response:
|
|
"""Returns info about the server."""
|
|
url = self.api_endpoint + api.ServerInfo.url.format(machine_id=machine_id)
|
|
response = self._httpx_client.get(url)
|
|
_handle_response(response)
|
|
response_object = api.ServerInfo.Response.parse_obj(response.json())
|
|
return response_object
|
|
|
|
def server_update(
|
|
self,
|
|
machine_id: str,
|
|
hostname: Union[str, None] = None,
|
|
autorenew: Union[bool, None] = None,
|
|
) -> None:
|
|
"""Update server settings."""
|
|
request = ServerUpdateRequest(hostname=hostname, autorenew=autorenew)
|
|
url = self.api_endpoint + f"/server/{machine_id}"
|
|
response = self._httpx_client.patch(url=url, json=request.dict())
|
|
_handle_response(response)
|
|
|
|
def servers_launched_from_token(
|
|
self, token: str
|
|
) -> api.ServersLaunchedFromToken.Response:
|
|
"""
|
|
Returns info of servers launched from a given token.
|
|
"""
|
|
url = self.api_endpoint + api.ServersLaunchedFromToken.url.format(token=token)
|
|
response = self._httpx_client.get(url)
|
|
_handle_response(response)
|
|
response_object = api.ServersLaunchedFromToken.Response.parse_obj(
|
|
response.json()
|
|
)
|
|
return response_object
|
|
|
|
def flavors(self) -> api.Flavors.Response:
|
|
"""Returns available flavors (server sizes)."""
|
|
url = self.api_endpoint + api.Flavors.url
|
|
response = self._httpx_client.get(url)
|
|
_handle_response(response)
|
|
response_object = api.Flavors.Response.parse_obj(response.json())
|
|
return response_object
|
|
|
|
def operating_systems(self) -> api.OperatingSystems.Response:
|
|
"""Returns available operating systems."""
|
|
url = self.api_endpoint + api.OperatingSystems.url
|
|
response = self._httpx_client.get(url)
|
|
_handle_response(response)
|
|
response_object = api.OperatingSystems.Response.parse_obj(response.json())
|
|
return response_object
|
|
|
|
def regions(self) -> api.Regions.Response:
|
|
"""Returns regions that you can launch a server in."""
|
|
url = self.api_endpoint + api.Regions.url
|
|
response = self._httpx_client.get(url)
|
|
_handle_response(response)
|
|
response_object = api.Regions.Response.parse_obj(response.json())
|
|
return response_object
|
|
|
|
def changelog(self) -> str:
|
|
"""Returns the API changelog."""
|
|
url = self.api_endpoint + "/changelog"
|
|
response = self._httpx_client.get(url)
|
|
_handle_response(response)
|
|
return response.text
|
|
|
|
def token_add(
|
|
self,
|
|
token: str,
|
|
dollars: int,
|
|
currency: Currency,
|
|
) -> api.TokenAdd.Response:
|
|
"""Add balance (money) to a token."""
|
|
url = self.api_endpoint + api.TokenAdd.url.format(token=token)
|
|
request = api.TokenAdd.Request(dollars=dollars, currency=currency)
|
|
response = self._httpx_client.post(url, json=request.dict())
|
|
_handle_response(response)
|
|
response_object = api.TokenAdd.Response.parse_obj(response.json())
|
|
return response_object
|
|
|
|
def token_balance(self, token: str) -> api.TokenBalance.Response:
|
|
"""Return a token's balance."""
|
|
url = self.api_endpoint + api.TokenBalance.url.format(token=token)
|
|
response = self._httpx_client.get(url)
|
|
_handle_response(response)
|
|
response_object = api.TokenBalance.Response.parse_obj(response.json())
|
|
return response_object
|
|
|
|
def token_info(self, token: str) -> TokenInfo:
|
|
"""Return information about a token, including balance."""
|
|
url = self.api_endpoint + f"/token/{token}/info"
|
|
response = self._httpx_client.get(url)
|
|
_handle_response(response)
|
|
response_object = TokenInfo.parse_obj(response.json())
|
|
return response_object
|
|
|
|
def token_get_messages(self, token: str) -> List[api.TokenMessage]:
|
|
"""Get messages for/from the token."""
|
|
url = self.api_endpoint + f"/token/{token}/messages"
|
|
response = self._httpx_client.get(url=url)
|
|
_handle_response(response)
|
|
|
|
return parse_obj_as(List[api.TokenMessage], response.json())
|
|
|
|
def token_send_message(self, token: str, message: str) -> None:
|
|
"""Send a message to SporeStack support."""
|
|
url = self.api_endpoint + f"/token/{token}/messages"
|
|
response = self._httpx_client.post(url=url, json={"message": message})
|
|
_handle_response(response)
|
|
|
|
def token_invoice(self, token: str, invoice: str) -> Invoice:
|
|
"""Get a particular invoice."""
|
|
url = self.api_endpoint + f"/token/{token}/invoices/{invoice}"
|
|
response = self._httpx_client.get(url=url)
|
|
_handle_response(response)
|
|
|
|
return parse_obj_as(Invoice, response.json())
|
|
|
|
def token_invoices(self, token: str) -> List[Invoice]:
|
|
"""Get token invoices."""
|
|
url = self.api_endpoint + f"/token/{token}/invoices"
|
|
response = self._httpx_client.get(url=url)
|
|
_handle_response(response)
|
|
|
|
return parse_obj_as(List[Invoice], response.json())
|