Spot.io API Calls via Python

spotinst_api.py

#!/opt/homebrew/bin/python3
"""
Perform Spot.io API calls with Python
Ensure body JSON is correct when performing API calls, otherwise the response will be 'bad request'.
"""

try:
    import requests
    import os
    import json
except ModuleNotFoundError:
    print("Please ensure proper modules are installed.\npip install requests")
    os._exit(1)

# Get Required Variables for Authentication
TOKEN = os.environ.get('SPOTINST_TOKEN')
ACCOUNT = os.environ.get('SPOTINST_ACCOUNT_AWS')

if not (TOKEN and ACCOUNT):
    print(f"Please set environment variables for token and account id.\n")
    os._exit(1)

class Endpoint:
    def __init__(self, url, body=None, account=ACCOUNT):
        """API Endpoint. Account and Token set by default
        
        Arguments:
        url: <class 'str'> representing URL endpoint of API
        body: <class 'str'> valid JSON dumped into a string
        """
        self.url = url
        self.body = body
        self.query_params = dict({TOKEN: account})
        self.headers = dict({
            'User-Agent': '_Custom_SDK_Agent',
            'Content-Type': 'application/json',
            'Authorization': 'Bearer ' + TOKEN
        })

    def _GET(self):
        """Send a GET request"""
        print(f"Sending GET request to {self.url}...\n")
        result = requests.get(self.url, params=self.query_params, headers=self.headers)

        return self.__HANDLE(result)

    def _POST(self):
        """Send a POST request"""
        print(f"Sending POST request to {self.url}...\n")
        result = requests.post(self.url, params=self.query_params, data=self.body, headers=self.headers)

        return self.__HANDLE(result)
    
    def _PUT(self):
        """Send a POST request"""
        print(f"Sending PUT request to {self.url}...\n")
        result = requests.put(self.url, params=self.query_params, data=self.body, headers=self.headers)

        return self.__HANDLE(result)

    def _DELETE(self):
        """Send a POST request"""
        print(f"Sending DELETE request to {self.url}...\n")
        result = requests.delete(self.url, params=self.query_params, headers=self.headers)

        return self.__HANDLE(result)
    
    @staticmethod
    def __HANDLE(result):
        """Handle response
        
        Returns: <class 'list'> of response items"""
        if result.status_code == requests.codes.ok:
            data = json.loads(result.content.decode('utf-8'))
            # print(data['response']['items'])
            return data['response']['items']

        else:
            print(f"Error: {result.reason}\n\n{result.text}\n")
            return 0


if __name__ == "__main__":

    """
    Reverse engineered Python SDK, as the official Python SDK is not updated to include full API functionality.
    Usage: Must write own API calls in the following format.

    1. Define base URL
    __base_api_url = 'URL'

    2. Define any required query parameters (usually account or cluster/group ID) [Optional]
    account_id = 'ID'

    3. Define body JSON [For PUT and POST requests]
    body = json.dumps(
        {
            key: value
        }
    )

    3. Instantiate Endpoint object with URL [and optionally] body data. I use API purpose as name of Endpoint object. Note if body is required, set as keyword argument
    get_users = Endpoint(f"{__base_api_url}/{account_id}, body=body)

    4. Call HTTP request method on Endpoint object
    get_users._GET()
    """

    # EXAMPLES BELOW

    # Get List of Users
    # __base_user_url = 'https://api.spotinst.io/setup/organization/user'
    # get_users = Endpoint(__base_user_url, account=ACCOUNT)._GET()

    # Get Ocean Cluster - GET
    # __base_ocean_url = 'https://api.spotinst.io/ocean/aws/k8s/cluster'
    # ocean_id = 'o-d109430a'
    # get_ocean_cluster = Endpoint(f"{__base_ocean_url}/{ocean_id}", account=ACCOUNT)
    # get_ocean_cluster._GET()

    # List Ocean Clusters - GET
    # __base_ocean_url = 'https://api.spotinst.io/ocean/aws/k8s/cluster'
    # account_id = 'act-9f3c19ea'
    # get_ocean_clusters = Endpoint(f"{__base_ocean_url}", account=ACCOUNT)
    # get_ocean_clusters._GET()

    # Update Ocean Cluster - PUT
    # __base_ocean_url = 'https://api.spotinst.io/ocean/aws/k8s/cluster'
    # account_id = 'act-9f3c19ea'
    # ocean_cluster_id = 'o-a39795cd'
    # body = json.dumps({"cluster": {
    #     "name": "freshworks-test-eks",
    #     "controllerClusterId": "freshworks-test-eks",
    #     "capacity": {
    #         "minimum": 0,
    #         "maximum": 500,
    #         "target": 1
    #     }
    # }})
    # update_ocean_cluster = Endpoint(f"{__base_ocean_url}/{ocean_cluster_id}", body=body, account=ACCOUNT)
    # update_ocean_cluster._PUT()

    # Initiate a Roll (Ocean) - POST
    # __base_roll_url = 'https://api.spotinst.io/ocean/aws/k8s/cluster'
    # ocean_cluster_id = 'o-a39795cd'
    # body = json.dumps({
    #     "roll": {
    #         "batchSizePercentage": 20,
    #         "comment": "Testing reverse engineered Python SDK",
    #         "respectPdb": True,
    #         "batchMinHealthyPercentage": 100
    #     }
    # })
    # initiate_roll = Endpoint(f"{__base_roll_url}/{ocean_cluster_id}/roll", body=body, account=ACCOUNT)
    # initiate_roll._POST()

    # Get Rightsizing Suggestions - POST
    # oceanClusterId = 'o-d109430a'
    # __base_rightsizing_url = f'https://api.spotinst.io/ocean/aws/k8s/cluster/{oceanClusterId}/rightSizing/suggestion'
    
    # get_suggestions = Endpoint(f"{__base_rightsizing_url}", account=ACCOUNT)
    # get_suggestions._POST()

    # Create Spot Account and Link an AWS Account to Spot
    # 1. Create Unlinked Spot Account
    # __base_account_url = 'https://api.spotinst.io/setup/account'
    # NAME = 'Karl-Testing-1'
    # body = json.dumps({
    # "account": {
    #     "name": f"{NAME}"
    #     }
    # })
    # create_account = Endpoint(f"{__base_account_url}", body=body, account=ACCOUNT)
    # spot_account_id = create_account._POST()[0]['id']
    
    # # 2. Create External ID: Change query param to newly created account ID
    # __base_account_link_aws_url = 'https://api.spotinst.io/setup/credentials/aws/externalId'
    # create_external_id = Endpoint(f"{__base_account_link_aws_url}", account=spot_account_id)
    # external_id = create_external_id._POST()[0]['externalId']

    # # 3. Use external_id and include in Spot Role CloudFormation stack (manual)
    # # TODO: Use boto3 to provision CFN stack
    # # Return the IAM Role ARN

    # # 4. Set Credentials for AWS
    # __base_account_credentials_aws_url = 'https://api.spotinst.io/setup/credentials/aws'
    # body = json.dumps({
    #     "credentials": {
    #         "iamRole": f"{IAM_ROLE_ARN}"
    #     }
    # })
    # create_aws_credentials = Endpoint(f"{__base_account_credentials_aws_url}", body=body, account=spot_account_id)
    # create_external_id._POST()