SporeStack
/
sporestack-python
2
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

WIP: Make APIClient, Token, and Server
ci/woodpecker/push/woodpecker Pipeline failed Details
ci/woodpecker/pr/woodpecker Pipeline failed Details

Browse Source
rewrite
SporeStack 6 months ago
parent e8a720e564
commit ad6fb4ce2e
2 changed files with 249 additions and 183 deletions
Show Stats Download Patch File Download Diff File

348
src/sporestack/api_client.py
Unescape Escape View File

@ -1,5 +1,6 @@
import logging
import os
from dataclasses import dataclass
from time import sleep
from typing import Any, Dict, Optional
@ -130,186 +131,167 @@ def _api_request(
raise Exception("Stuff broke strangely. Please contact SporeStack support.")
def launch(
machine_id: str,
days: int,
flavor: str,
operating_system: str,
ssh_key: str,
token: str,
api_endpoint: str = API_ENDPOINT,
region: Optional[str] = None,
retry: bool = False,
quote: bool = False,
hostname: str = "",
autorenew: bool = False,
) -> api.ServerLaunch.Response:
request = api.ServerLaunch.Request(
days=days,
token=token,
flavor=flavor,
region=region,
operating_system=operating_system,
ssh_key=ssh_key,
quote=quote,
hostname=hostname,
autorenew=autorenew,
)
url = api_endpoint + api.ServerLaunch.url.format(machine_id=machine_id)
response = _api_request(url=url, json_params=request.dict(), retry=retry)
response_object = api.ServerLaunch.Response.parse_obj(response)
assert response_object.machine_id == machine_id
return response_object
def topup(
machine_id: str,
days: int,
token: str,
api_endpoint: str = API_ENDPOINT,
retry: bool = False,
) -> api.ServerTopup.Response:
"""
Topup a server.
"""
request = api.ServerTopup.Request(days=days, token=token)
url = api_endpoint + api.ServerTopup.url.format(machine_id=machine_id)
response = _api_request(url=url, json_params=request.dict(), retry=retry)
response_object = api.ServerTopup.Response.parse_obj(response)
assert response_object.machine_id == machine_id
return response_object
def autorenew_enable(machine_id: str, api_endpoint: str = API_ENDPOINT) -> None:
"""
Enable autorenew on a server.
"""
url = api_endpoint + api.ServerEnableAutorenew.url.format(machine_id=machine_id)
_api_request(url, empty_post=True)
def autorenew_disable(machine_id: str, api_endpoint: str = API_ENDPOINT) -> None:
"""
Disable autorenew on a server.
"""
url = api_endpoint + api.ServerDisableAutorenew.url.format(machine_id=machine_id)
_api_request(url, empty_post=True)
def start(machine_id: str, api_endpoint: str = API_ENDPOINT) -> None:
"""
Boots the server.
"""
url = api_endpoint + api.ServerStart.url.format(machine_id=machine_id)
_api_request(url, empty_post=True)
def stop(machine_id: str, api_endpoint: str = API_ENDPOINT) -> None:
"""
Powers off the server.
"""
url = api_endpoint + api.ServerStop.url.format(machine_id=machine_id)
_api_request(url, empty_post=True)
def destroy(machine_id: str, api_endpoint: str = API_ENDPOINT) -> None:
"""
Destroys the server.
"""
url = api_endpoint + api.ServerDestroy.url.format(machine_id=machine_id)
_api_request(url, empty_post=True)
def delete(machine_id: str, api_endpoint: str = API_ENDPOINT) -> None:
"""
Deletes the server. (Deprecated, use destroy instead)
"""
destroy(machine_id, api_endpoint)
def forget(machine_id: str, api_endpoint: str = API_ENDPOINT) -> None:
"""
Forget about a destroyed/deleted server.
"""
url = api_endpoint + api.ServerForget.url.format(machine_id=machine_id)
_api_request(url, empty_post=True)
def rebuild(machine_id: str, api_endpoint: str = API_ENDPOINT) -> None:
"""
Rebuilds the server with the operating system and SSH key set at launch time.
Deletes all of the data on the server!
"""
url = api_endpoint + api.ServerRebuild.url.format(machine_id=machine_id)
_api_request(url, empty_post=True)
def info(machine_id: str, api_endpoint: str = API_ENDPOINT) -> api.ServerInfo.Response:
"""
Returns info about the server.
"""
url = api_endpoint + api.ServerInfo.url.format(machine_id=machine_id)
response = _api_request(url)
response_object = api.ServerInfo.Response.parse_obj(response)
assert response_object.machine_id == machine_id
return response_object
def servers_launched_from_token(
token: str, api_endpoint: str = API_ENDPOINT
) -> api.ServersLaunchedFromToken.Response:
"""
Returns info of servers launched from a given token.
"""
url = api_endpoint + api.ServersLaunchedFromToken.url.format(token=token)
response = _api_request(url)
response_object = api.ServersLaunchedFromToken.Response.parse_obj(response)
return response_object
def flavors(api_endpoint: str = API_ENDPOINT) -> api.Flavors.Response:
"""
Returns available flavors.
"""
url = api_endpoint + api.Flavors.url
response = _api_request(url)
response_object = api.Flavors.Response.parse_obj(response)
return response_object
def operating_systems(
api_endpoint: str = API_ENDPOINT,
) -> api.OperatingSystems.Response:
"""
Returns available operating systems.
"""
url = api_endpoint + api.OperatingSystems.url
response = _api_request(url)
response_object = api.OperatingSystems.Response.parse_obj(response)
return response_object
def token_add(
token: str,
dollars: int,
currency: str,
api_endpoint: str = API_ENDPOINT,
retry: bool = False,
) -> api.TokenAdd.Response:
request = api.TokenAdd.Request(dollars=dollars, currency=currency)
url = api_endpoint + api.TokenAdd.url.format(token=token)
response = _api_request(url=url, json_params=request.dict(), retry=retry)
response_object = api.TokenAdd.Response.parse_obj(response)
assert response_object.token == token
return response_object
def token_balance(
token: str, api_endpoint: str = API_ENDPOINT
) -> api.TokenBalance.Response:
url = api_endpoint + api.TokenBalance.url.format(token=token)
response = _api_request(url=url)
response_object = api.TokenBalance.Response.parse_obj(response)
assert response_object.token == token
return response_object
@dataclass
class APIClient:
api_endpoint: str = API_ENDPOINT
def launch(
self,
machine_id: str,
days: int,
flavor: str,
operating_system: str,
ssh_key: str,
token: str,
region: Optional[str] = None,
quote: bool = False,
hostname: str = "",
autorenew: bool = False,
) -> api.ServerLaunch.Response:
request = api.ServerLaunch.Request(
days=days,
token=token,
flavor=flavor,
region=region,
operating_system=operating_system,
ssh_key=ssh_key,
quote=quote,
hostname=hostname,
autorenew=autorenew,
)
url = self.api_endpoint + api.ServerLaunch.url.format(machine_id=machine_id)
response = _api_request(url=url, json_params=request.dict())
response_object = api.ServerLaunch.Response.parse_obj(response)
assert response_object.machine_id == machine_id
return response_object
def server_topup(
self,
machine_id: str,
days: int,
token: str,
) -> api.ServerTopup.Response:
"""
Topup a server.
"""
request = api.ServerTopup.Request(days=days, token=token)
url = self.api_endpoint + api.ServerTopup.url.format(machine_id=machine_id)
response = _api_request(url=url, json_params=request.dict())
response_object = api.ServerTopup.Response.parse_obj(response)
assert response_object.machine_id == machine_id
return response_object
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
)
_api_request(url, empty_post=True)
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
)
_api_request(url, empty_post=True)
def sever_start(self, machine_id: str) -> None:
"""
Boots the server.
"""
url = self.api_endpoint + api.ServerStart.url.format(machine_id=machine_id)
_api_request(url, empty_post=True)
def server_stop(self, machine_id: str) -> None:
"""
Powers off the server.
"""
url = self.api_endpoint + api.ServerStop.url.format(machine_id=machine_id)
_api_request(url, empty_post=True)
def server_delete(self, machine_id: str) -> None:
"""
Deletes the server.
"""
url = self.api_endpoint + api.ServerDelete.url.format(machine_id=machine_id)
_api_request(url, empty_post=True)
def server_forget(self, machine_id: str) -> None:
"""
Forget about a destroyed/deleted server.
"""
url = self.api_endpoint + api.ServerForget.url.format(machine_id=machine_id)
_api_request(url, empty_post=True)
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)
_api_request(url, empty_post=True)
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 = _api_request(url)
response_object = api.ServerInfo.Response.parse_obj(response)
assert response_object.machine_id == machine_id
return response_object
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 = _api_request(url)
response_object = api.ServersLaunchedFromToken.Response.parse_obj(response)
return response_object
def flavors(self) -> api.Flavors.Response:
"""
Returns available flavors.
"""
url = self.api_endpoint + api.Flavors.url
response = _api_request(url)
response_object = api.Flavors.Response.parse_obj(response)
return response_object
def operating_systems(self) -> api.OperatingSystems.Response:
"""
Returns available operating systems.
"""
url = self.api_endpoint + api.OperatingSystems.url
response = _api_request(url)
response_object = api.OperatingSystems.Response.parse_obj(response)
return response_object
def token_add(
self,
token: str,
dollars: int,
currency: str,
retry: bool = False,
) -> api.TokenAdd.Response:
request = api.TokenAdd.Request(dollars=dollars, currency=currency)
url = self.api_endpoint + api.TokenAdd.url.format(token=token)
response = _api_request(url=url, json_params=request.dict(), retry=retry)
response_object = api.TokenAdd.Response.parse_obj(response)
assert response_object.token == token
return response_object
def token_balance(self, token: str) -> api.TokenBalance.Response:
url = self.api_endpoint + api.TokenBalance.url.format(token=token)
response = _api_request(url=url)
response_object = api.TokenBalance.Response.parse_obj(response)
assert response_object.token == token
return response_object

84
src/sporestack/client.py
Unescape Escape View File

@ -0,0 +1,84 @@
from . import api_client
from .utils import random_machine_id, random_token
@dataclass
class Server:
machine_id: str
api_client: api_client.APIClient = api_client.APIClient()
def info(self) -> api.ServerInfo.Response:
return api_client.server_info(self.machine_id)
def rebuild(self) -> None:
api_client.server_rebuild(self.machine_id)
def forget(self) -> None:
api_client.server_forget(self.machine_id)
def delete(self) -> None:
api_client.server_delete(self.machine_id)
def start(self) -> None:
"""Powers on the server."""
api_client.server_start(self.machine_id)
def stop(self) -> None:
"""Powers off the server."""
api_client.server_stop(self.machine_id)
def autorenew_enable(self) -> None:
"""Enables autorenew on the server."""
api_client.autorenew_enable(self.machine_id)
def autorenew_disable(self) -> None:
"""Disables autorenew on the server."""
api_client.autorenew_disable(self.machine_id)
def topup(days: int, token: str) -> None:
"""Renew the server for the amount of days specified, from the token specified."""
api_client.server_topup(machine_id=self.machine_id, days=days, token=token)
@dataclass
class Token:
token: str = random_token()
api_client: api_client.APIClient = api_client.APIClient()
def add(self, dollars: int, currency: str) -> None:
"""FIXME"""
self.api_client.token_add(token=token, dollars=dollars, currency=currency)
def balance(self) -> int:
"""Returns the token's balance in cents."""
self.api_client.token_balance(token=token).cents
def servers(self) -> List[Server]:
server_classes = []
for server in api_client.servers_launched_from_token():
server_classes.append(
Server(machine_id=server.machine_id, api_client=self.api_client)
)
return server_classes
def launch_server(
self,
ssh_key: str,
flavor: str,
days: int,
region: Union[str, None] = None,
hostname: str = "",
autorenew: bool = False,
machine_id=random_machine_id(),
) -> Server:
self.api_client.server_launch(
machine_id=machine_id,
days=days,
token=self.token,
region=region,
operating_system=operating_system,
ssh_key=ssh_key,
hostname=hostname,
autorenew=autorenew,
)
return Server(machine_id=machine_id, api_client=self.api_client)
Write Preview
Loading…
Cancel
Save