diff --git a/src/sporestack/api_client.py b/src/sporestack/api_client.py index 932b5a8..0334c3f 100644 --- a/src/sporestack/api_client.py +++ b/src/sporestack/api_client.py @@ -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 +@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 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 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_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(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 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 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 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 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. -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) + 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 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 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 delete(machine_id: str, api_endpoint: str = API_ENDPOINT) -> None: - """ - Deletes the server. (Deprecated, use destroy instead) - """ - destroy(machine_id, api_endpoint) + 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 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 + 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 diff --git a/src/sporestack/client.py b/src/sporestack/client.py new file mode 100644 index 0000000..7db367f --- /dev/null +++ b/src/sporestack/client.py @@ -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)