aws_ec2_pricing.py

import requests
import threading
import json
from json import JSONDecodeError


class AwsEc2Price:
    def __init__(self, term_offering_class_, term_length_, payment_option_, region_, operating_system_, tenancy_):
        self.term_offering_class = term_offering_class_
        self.term_length = term_length_
        self.payment_option = payment_option_
        self.region = region_
        self.operating_system = operating_system_
        self.tenancy = tenancy_

    def get_ec2_prices(self):
        response = requests.get(
            f"https://a0.p.awsstatic.com/pricing/1.0/ec2/region/{self.region}/reserved-instance/linux/index.json"
        )

        prices_list = []

        try:
            pricing_data = response.json()
            prices = pricing_data["prices"]

            for price in prices:
                if (price["attributes"]["aws:offerTermOfferingClass"] == self.term_offering_class and
                        price["attributes"]["aws:offerTermLeaseLength"] == self.term_length and
                        price["attributes"]["aws:offerTermPurchaseOption"] == self.payment_option and
                        price["attributes"]["aws:ec2:operatingSystem"] == self.operating_system and
                        price["attributes"]["aws:ec2:tenancy"] == self.tenancy):
                    prices_list.append(price)
        except JSONDecodeError:
            print("JSON Decode error")

        return prices_list

    @staticmethod
    def save_prices_in_file(prices_list):
        prices = json.dumps(prices_list)
        with open('./prices/ec2_price.json', 'w') as prices_file:
            prices_file.write(prices)

    # Running get_ec2_prices and save_prices_in_file function using this function
    # This will enable us to run both functions in a single thread as both functions are dependent on each other
    def get_and_save_ec2_prices(self):
        prices_list = self.get_ec2_prices()
        self.save_prices_in_file(prices_list)


# Get the details from user
term_offering_class = input("Enter the term offering class (standard / convertible): ")
term_length = input("Enter the term length: ")
payment_option = input("Enter the payment option: ")
region = input("Enter the region: ")
operating_system = input("Enter the operating system: ")
tenancy = input("Enter the tenancy: ")

ec2_price = AwsEc2Price(term_offering_class, term_length, payment_option, region, operating_system, tenancy)

get_and_save_ec2_prices_in_file = threading.Thread(target=ec2_price.get_and_save_ec2_prices)
get_and_save_ec2_prices_in_file.start()