_aws

sc_crawler.vendors._aws

Functions:

Name	Description
inventory_compliance_frameworks	Manual list of compliance frameworks known for AWS.
inventory_regions	List all available AWS regions via boto3 calls.
inventory_zones	List all available AWS availability zones via boto3 calls.
inventory_servers	List all available AWS instance types in all regions via boto3 calls.
inventory_server_prices	List all on-demand instance prices in all regions via boto3 calls.
inventory_server_prices_spot	List all spot instance prices in all availability zones via boto3 calls.
inventory_storages	List all storage types via boto3 calls.
inventory_storage_prices	List all storage prices in all regions via boto3 calls.
inventory_traffic_prices	List all inbound and outbound traffic prices in all regions via boto3 calls.
inventory_ipv4_prices	List IPV4 prices in all regions via boto3 calls.
inventory_databases	List all available AWS RDS PostgreSQL instance types via boto3 calls.
inventory_database_prices	List all on-demand AWS RDS PostgreSQL prices in all regions via boto3 calls.
inventory_database_storages	List all AWS RDS PostgreSQL storage types via boto3 calls.
inventory_database_storage_prices	List all AWS RDS PostgreSQL storage prices in all regions via boto3 calls.

inventory_compliance_frameworks

inventory_compliance_frameworks(vendor)

Manual list of compliance frameworks known for AWS.

Resources: https://aws.amazon.com/compliance/programs/

Source code in sc_crawler/vendors/_aws.py

def inventory_compliance_frameworks(vendor):
    """Manual list of compliance frameworks known for AWS.

    Resources: <https://aws.amazon.com/compliance/programs/>
    """
    return map_compliance_frameworks_to_vendor(
        vendor.vendor_id, ["hipaa", "soc2t2", "iso27001"]
    )

inventory_regions

inventory_regions(vendor)

List all available AWS regions via boto3 calls.

Some data sources are not available from APIs, and were collected manually:

• launch date: https://aws.amazon.com/about-aws/global-infrastructure/regions_az/,
• energy source: https://sustainability.aboutamazon.com/products-services/the-cloud?energyType=true#renewable-energy and https://www.aboutamazon.com/news/sustainability/amazon-renewable-energy-goal,
• lon/lat coordinates: https://gist.github.com/martinheidegger/88950cb51ee5bdeafd51bc55287b1092 and approximation based on the city when no more accurate data was available.

Source code in sc_crawler/vendors/_aws.py

def inventory_regions(vendor):
    """List all available AWS regions via `boto3` calls.

    Some data sources are not available from APIs, and were collected manually:

    - launch date: <https://aws.amazon.com/about-aws/global-infrastructure/regions_az/>,
    - energy source: <https://sustainability.aboutamazon.com/products-services/the-cloud?energyType=true#renewable-energy> and <https://www.aboutamazon.com/news/sustainability/amazon-renewable-energy-goal>,
    - lon/lat coordinates: <https://gist.github.com/martinheidegger/88950cb51ee5bdeafd51bc55287b1092> and approximation based on the city when no more accurate data was available.
    """  # noqa: E501
    regions = [
        {
            "region_id": "af-south-1",
            "name": "Africa (Cape Town)",
            "vendor_id": vendor.vendor_id,
            "country_id": "ZA",
            "city": "Cape Town",
            "founding_year": 2020,
            "lat": -33.914651,
            "lon": 18.3758801,
        },
        {
            "region_id": "ap-east-1",
            "name": "Asia Pacific (Hong Kong)",
            "vendor_id": vendor.vendor_id,
            "country_id": "HK",
            "city": "Hong Kong",
            "founding_year": 2019,
            "lat": 22.2908475,
            "lon": 114.2723379,
        },
        {
            "region_id": "ap-east-2",
            "name": "Asia Pacific (Taipei)",
            "vendor_id": vendor.vendor_id,
            "country_id": "TW",
            "city": "Taipei",
            "founding_year": 2025,
            # approximation based on city
            "lat": 25.037518,
            "lon": 121.563667,
        },
        {
            "region_id": "ap-northeast-1",
            "name": "Asia Pacific (Tokyo)",
            "vendor_id": vendor.vendor_id,
            "country_id": "JP",
            "city": "Tokyo",
            "founding_year": 2011,
            "lat": 35.617436,
            "lon": 139.7459176,
        },
        {
            "region_id": "ap-northeast-2",
            "name": "Asia Pacific (Seoul)",
            "vendor_id": vendor.vendor_id,
            "country_id": "KR",
            "city": "Seoul",
            "founding_year": 2016,
            "lat": 37.5616592,
            "lon": 126.8736237,
        },
        {
            "region_id": "ap-northeast-3",
            "name": "Asia Pacific (Osaka)",
            "vendor_id": vendor.vendor_id,
            "country_id": "JP",
            "city": "Osaka",
            "founding_year": 2021,
            "lat": 34.693889,
            "lon": 135.502222,
        },
        {
            "region_id": "ap-south-1",
            "name": "Asia Pacific (Mumbai)",
            "vendor_id": vendor.vendor_id,
            "country_id": "IN",
            "city": "Mumbai",
            "founding_year": 2016,
            "lat": 19.2425503,
            "lon": 72.9667878,
        },
        {
            "region_id": "ap-south-2",
            "name": "Asia Pacific (Hyderabad)",
            "vendor_id": vendor.vendor_id,
            "country_id": "IN",
            "city": "Hyderabad",
            "founding_year": 2022,
            # approximation based on city location
            "lat": 17.412281,
            "lon": 78.243237,
        },
        {
            "region_id": "ap-southeast-1",
            "name": "Asia Pacific (Singapore)",
            "vendor_id": vendor.vendor_id,
            "country_id": "SG",
            "city": "Singapore",
            "founding_year": 2010,
            "lat": 1.3218269,
            "lon": 103.6930643,
        },
        {
            "region_id": "ap-southeast-2",
            "name": "Asia Pacific (Sydney)",
            "vendor_id": vendor.vendor_id,
            "country_id": "AU",
            "city": "Sydney",
            "founding_year": 2012,
            "lat": -33.9117717,
            "lon": 151.1907535,
        },
        {
            "region_id": "ap-southeast-3",
            "name": "Asia Pacific (Jakarta)",
            "vendor_id": vendor.vendor_id,
            "country_id": "ID",
            "city": "Jakarta",
            "founding_year": 2021,
            "lat": -6.2,
            "lon": 106.816667,
        },
        {
            "region_id": "ap-southeast-4",
            "name": "Asia Pacific (Melbourne)",
            "vendor_id": vendor.vendor_id,
            "country_id": "AU",
            "city": "Melbourne",
            "founding_year": 2023,
            # approximation based on city location
            "lat": -37.8038607,
            "lon": 144.7119569,
        },
        {
            "region_id": "ap-southeast-5",
            "name": "Asia Pacific (Malaysia)",
            "vendor_id": vendor.vendor_id,
            "country_id": "MY",
            "founding_year": 2024,
            # approximation based on country
            "lat": 4.1230237,
            "lon": 104.3228082,
        },
        {
            "region_id": "ap-southeast-6",
            "name": "Asia Pacific (New Zealand)",
            "vendor_id": vendor.vendor_id,
            "country_id": "NZ",
            "founding_year": 2025,
            # approximation based on country
            "lat": -40.900775,
            "lon": 174.802185,
        },
        {
            "region_id": "ap-southeast-7",
            "name": "Asia Pacific (Thailand)",
            "vendor_id": vendor.vendor_id,
            "country_id": "TH",
            "founding_year": 2022,
            # approximation based on country
            "lat": 15.870032,
            "lon": 100.992538,
        },
        {
            "region_id": "ca-central-1",
            "name": "Canada (Central)",
            "vendor_id": vendor.vendor_id,
            "country_id": "CA",
            "city": "Quebec",  # NOTE needs city name
            "founding_year": 2016,
            "lat": 45.5,
            "lon": -73.6,
        },
        {
            "region_id": "ca-west-1",
            "name": "Canada West (Calgary)",
            "vendor_id": vendor.vendor_id,
            "country_id": "CA",
            "city": "Calgary",
            "founding_year": 2023,
            # approximation based on city location
            "lat": 51.046574,
            "lon": -114.129024,
        },
        {
            "region_id": "cn-north-1",
            "name": "China (Beijing)",
            "vendor_id": vendor.vendor_id,
            "country_id": "CN",
            "city": "Beijing",
            "founding_year": 2016,
            "lat": 39.8094478,
            "lon": 116.5783234,
        },
        {
            "region_id": "cn-northwest-1",
            "name": "China (Ningxia)",
            "vendor_id": vendor.vendor_id,
            "country_id": "CN",
            "city": "Ningxia",  # NOTE needs city name
            "founding_year": 2017,
            "lat": 37.5024418,
            "lon": 105.1627193,
        },
        {
            "region_id": "eu-central-1",
            "name": "Europe (Frankfurt)",
            "aliases": ["EU (Frankfurt)"],
            "vendor_id": vendor.vendor_id,
            "country_id": "DE",
            "city": "Frankfurt",
            "founding_year": 2014,
            "lat": 50.0992094,
            "lon": 8.6303932,
        },
        {
            "region_id": "eu-central-2",
            "name": "Europe (Zurich)",
            "vendor_id": vendor.vendor_id,
            "country_id": "CH",
            "city": "Zurich",
            "founding_year": 2022,
            # approximation based on city location
            "lat": 47.3862924,
            "lon": 8.4448814,
        },
        {
            "region_id": "eu-north-1",
            "name": "Europe (Stockholm)",
            "aliases": ["EU (Stockholm)"],
            "vendor_id": vendor.vendor_id,
            "country_id": "SE",
            "city": "Stockholm",
            "founding_year": 2018,
            "lat": 59.326242,
            "lon": 17.8419717,
        },
        {
            "region_id": "eu-south-1",
            "name": "Europe (Milan)",
            "aliases": ["EU (Milan)"],
            "vendor_id": vendor.vendor_id,
            "country_id": "IT",
            "city": "Milan",
            "founding_year": 2020,
            "lat": 45.4628328,
            "lon": 9.1076927,
        },
        {
            "region_id": "eu-south-2",
            "name": "Europe (Spain)",
            "vendor_id": vendor.vendor_id,
            "country_id": "ES",
            "city": "Aragón",  # NOTE needs city name
            "founding_year": 2022,
            # approximation based on city location
            "lat": 41.7943702,
            "lon": -0.8516735,
        },
        {
            "region_id": "eu-west-1",
            "name": "Europe (Ireland)",
            "aliases": ["EU (Ireland)"],
            "vendor_id": vendor.vendor_id,
            "country_id": "IE",
            "city": "Dublin",
            "founding_year": 2007,
            "lat": 53.4056545,
            "lon": -6.224503,
        },
        {
            "region_id": "eu-west-2",
            "name": "Europe (London)",
            "aliases": ["EU (London)"],
            "vendor_id": vendor.vendor_id,
            "country_id": "GB",
            "city": "London",
            "founding_year": 2016,
            "lat": 51.5085036,
            "lon": -0.0609266,
        },
        {
            "region_id": "eu-west-3",
            "name": "Europe (Paris)",
            "aliases": ["EU (Paris)"],
            "vendor_id": vendor.vendor_id,
            "country_id": "FR",
            "city": "Paris",
            "founding_year": 2017,
            "lat": 48.6009709,
            "lon": 2.2976644,
        },
        {
            "region_id": "il-central-1",
            "name": "Israel (Tel Aviv)",
            "vendor_id": vendor.vendor_id,
            "country_id": "IL",
            "city": "Tel Aviv",
            "founding_year": 2023,
            # approximation based on city location
            "lat": 32.0491183,
            "lon": 34.7891105,
        },
        {
            "region_id": "me-central-1",
            "name": "Middle East (UAE)",
            "vendor_id": vendor.vendor_id,
            "country_id": "AE",
            # NOTE city and state unknown
            "display_name": "United Arab Emirates",
            "founding_year": 2022,
            # approximation based on country
            "lat": 25.0647937,
            "lon": 55.1363688,
        },
        {
            "region_id": "me-south-1",
            "name": "Middle East (Bahrain)",
            "vendor_id": vendor.vendor_id,
            "country_id": "BH",
            # NOTE city and stateunknown
            "display_name": "Bahrain",
            "founding_year": 2019,
            "lat": 25.941298,
            "lon": 50.3073907,
        },
        {
            "region_id": "mx-central-1",
            "name": "Mexico (Central)",
            "vendor_id": vendor.vendor_id,
            "country_id": "MX",
            "founding_year": 2025,
            # approximation based on country
            "lat": 20.5896,
            "lon": -100.3897,
        },
        {
            "region_id": "sa-east-1",
            "name": "South America (Sao Paulo)",
            "vendor_id": vendor.vendor_id,
            "country_id": "BR",
            "city": "Sao Paulo",
            "founding_year": 2011,
            "lat": -23.4925798,
            "lon": -46.8105593,
        },
        {
            "region_id": "us-east-1",
            "name": "US East (N. Virginia)",
            "vendor_id": vendor.vendor_id,
            "country_id": "US",
            "state": "Northern Virgina",
            # NOTE city unknown
            "founding_year": 2006,
            "lat": 38.9940541,
            "lon": -77.4524237,
        },
        {
            "region_id": "us-east-2",
            "name": "US East (Ohio)",
            "vendor_id": vendor.vendor_id,
            "country_id": "US",
            "state": "Ohio",
            # NOTE city unknown
            "founding_year": 2016,
            "lat": 40.0946354,
            "lon": -82.7541337,
        },
        {
            "region_id": "us-west-1",
            "name": "US West (N. California)",
            "vendor_id": vendor.vendor_id,
            "country_id": "US",
            "state": "California",
            # NOTE city unknown
            "founding_year": 2009,
            "lat": 37.443680,
            "lon": -122.153664,
        },
        {
            "region_id": "us-west-2",
            "name": "US West (Oregon)",
            "vendor_id": vendor.vendor_id,
            "country_id": "US",
            "state": "Oregon",
            # NOTE city unknown
            "founding_year": 2011,
            "lat": 45.9174667,
            "lon": -119.2684488,
        },
    ]

    # add API reference and display names
    for region in regions:
        region["api_reference"] = region["region_id"]
        if region.get("display_name") is None:
            display_name_prefix = region.get("city", region.get("state", ""))
            region["display_name"] = f"{display_name_prefix} ({region['country_id']})"

    # look for undocumented (new) regions in AWS
    supported_regions = [d["region_id"] for d in regions]
    available_regions = _boto_describe_regions()
    for available_region in available_regions:
        with sentry_capture_or_raise(vendor=vendor):
            region_name = available_region["RegionName"]
            if "gov" in region_name:
                continue
            if region_name not in supported_regions:
                raise NotImplementedError(f"Unsupported AWS region: {region_name}")

    # mark inactive regions
    active_regions = [region["RegionName"] for region in available_regions]
    for region in regions:
        if region["region_id"] in active_regions:
            region["status"] = "active"
        else:
            region["status"] = "inactive"

    # all regions are matched with 100% renewable energy
    # https://www.aboutamazon.com/news/sustainability/amazon-renewable-energy-goal
    for region in regions:
        region["green_energy"] = True

    return regions

inventory_zones

inventory_zones(vendor)

List all available AWS availability zones via boto3 calls.

Source code in sc_crawler/vendors/_aws.py

def inventory_zones(vendor):
    """List all available AWS availability zones via `boto3` calls."""
    vendor.progress_tracker.start_task(
        name="Scanning region(s) for zone(s)", total=len(vendor.regions)
    )

    def get_zones(region: Region, vendor: Vendor) -> List[dict]:
        new = []
        if region.status == "active":

            def on_error():
                region.status = Status.INACTIVE
                vendor.log(
                    f"Marking region {region.region_id} as inactive due to error "
                    "while fetching availability zones.",
                    WARN,
                )

            with sentry_capture_or_raise(vendor=vendor, on_error=on_error):
                for zone in _boto_describe_availability_zones(region.region_id):
                    new.append(
                        {
                            "zone_id": zone["ZoneId"],
                            "name": zone["ZoneName"],
                            "api_reference": zone["ZoneName"],
                            "display_name": zone["ZoneName"],
                            "region_id": region.region_id,
                            "vendor_id": vendor.vendor_id,
                        }
                    )

        vendor.progress_tracker.advance_task()
        return new

    with ThreadPoolExecutor(max_workers=8) as executor:
        zones = executor.map(get_zones, vendor.regions, repeat(vendor))
    zones = list(chain.from_iterable(zones))
    vendor.progress_tracker.hide_task()
    return zones

inventory_servers

inventory_servers(vendor)

List all available AWS instance types in all regions via boto3 calls.

Source code in sc_crawler/vendors/_aws.py

def inventory_servers(vendor):
    """List all available AWS instance types in all regions via `boto3` calls."""
    # TODO consider dropping this in favor of pricing.get_products, as
    #      it has info e.g. on instanceFamily although other fields
    #      are messier (e.g. extract memory from string)
    servers = parallel_fetch_servers(
        vendor, _boto_describe_instance_types, "InstanceType", "regions"
    )
    servers = preprocess_servers(servers, vendor, _make_server_from_instance_type)
    return servers

inventory_server_prices

inventory_server_prices(vendor)

List all on-demand instance prices in all regions via boto3 calls.

Source code in sc_crawler/vendors/_aws.py

def inventory_server_prices(vendor):
    """List all on-demand instance prices in all regions via `boto3` calls."""
    vendor.progress_tracker.start_task(
        name="Searching for ondemand server_price(s)", total=None
    )
    products = _boto_get_products(
        service_code="AmazonEC2",
        filters={
            # TODO ingest win, mac etc others
            "operatingSystem": "Linux",
            "preInstalledSw": "NA",
            "licenseModel": "No License required",
            "locationType": "AWS Region",
            "capacitystatus": "Used",
            # TODO reserved pricing options - might decide not to, as not in scope?
            "marketoption": "OnDemand",
            # TODO dedicated options?
            "tenancy": "Shared",
        },
    )
    vendor.progress_tracker.hide_task()

    # lookup tables
    regions = scmodels_to_dict(vendor.regions, keys=["name", "aliases"])
    servers = scmodels_to_dict(vendor.servers, keys=["server_id"])

    # check all regions for instance types per zone
    active_region_ids = [
        r.api_reference for r in regions.values() if r.status == Status.ACTIVE
    ]
    vendor.progress_tracker.start_task(
        name="Look up supported server types in all ACTIVE regions/zones",
        total=len(active_region_ids),
    )
    with ThreadPoolExecutor(max_workers=8) as executor:
        regions_servers = executor.map(
            _describe_instance_type_offerings_per_zone_with_progress,
            active_region_ids,
            repeat(vendor),
        )
    regions_servers = dict(zip(active_region_ids, regions_servers))
    vendor.progress_tracker.hide_task()

    server_prices = []
    vendor.progress_tracker.start_task(
        name="Preprocess ondemand server_price(s)", total=len(products)
    )
    for product in products:
        try:
            attributes = product["product"]["attributes"]
            # early drop Gov regions
            if "GovCloud" in attributes["location"]:
                continue
            server = servers[attributes["instanceType"]]
            region = regions[attributes["location"]]
            zones = regions_servers.get(region.api_reference, {}).get(
                server.server_id, []
            )
            price = _extract_ondemand_price(product["terms"])
            for zone in zones:
                server_prices.append(
                    {
                        "vendor_id": vendor.vendor_id,
                        "region_id": region.region_id,
                        "zone_id": zone,
                        "server_id": server.server_id,
                        # TODO ingest other OSs
                        "operating_system": "Linux",
                        "allocation": Allocation.ONDEMAND,
                        "price": float(price[0]),
                        "currency": price[1],
                        "unit": PriceUnit.HOUR,
                    }
                )
        except KeyError as e:
            vendor.log(
                f"Cannot make ondemand server_price due to unknown {str(e)}: {str(attributes)}",
                DEBUG,
            )
        finally:
            vendor.progress_tracker.advance_task()
    vendor.progress_tracker.hide_task()
    return server_prices

inventory_server_prices_spot

inventory_server_prices_spot(vendor)

List all spot instance prices in all availability zones via boto3 calls.

Source code in sc_crawler/vendors/_aws.py

def inventory_server_prices_spot(vendor):
    """List all spot instance prices in all availability zones via `boto3` calls."""
    vendor.progress_tracker.start_task(
        name="Scanning regions for spot server_price(s)",
        total=len(vendor.regions),
    )

    def get_spot_prices(region: Region, vendor: Vendor) -> List[dict]:
        new = []
        if region.status == "active":
            try:
                new = _describe_spot_price_history(region.region_id)
                vendor.log(
                    f"{len(new)} spot server_price(s) found in {region.region_id}."
                )
            except ClientError as e:
                vendor.log(
                    f"Cannot get spot server_price in {region.region_id}: {str(e)}"
                )
        vendor.progress_tracker.advance_task()
        return new

    with ThreadPoolExecutor(max_workers=8) as executor:
        products = executor.map(get_spot_prices, vendor.regions, repeat(vendor))
    products = list(chain.from_iterable(products))
    vendor.log(f"{len(products)} spot server_price(s) found.")
    vendor.progress_tracker.hide_task()

    # lookup tables
    zones = scmodels_to_dict(vendor.zones, keys=["name"])
    servers = scmodels_to_dict(vendor.servers, keys=["server_id"])

    server_prices = []
    vendor.progress_tracker.start_task(
        name="Preprocess spot server_price(s)", total=len(products)
    )
    for product in products:
        try:
            zone = zones[product["AvailabilityZone"]]
            server = servers[product["InstanceType"]]
        except KeyError as e:
            vendor.log(
                f"Cannot make ondemand server_price due to unknown {str(e)}: {str(product)}",
                DEBUG,
            )
            continue
        server_prices.append(
            {
                "vendor_id": vendor.vendor_id,
                "region_id": zone.region.region_id,
                "zone_id": zone.zone_id,
                "server_id": server.server_id,
                # TODO ingest other OSs
                "operating_system": "Linux",
                "allocation": Allocation.SPOT,
                "price": float(product["SpotPrice"]),
                "currency": "USD",
                "unit": PriceUnit.HOUR,
                # use reported time instead of current timestamp
                "observed_at": product["Timestamp"],
            }
        )
        vendor.progress_tracker.advance_task()
    vendor.progress_tracker.hide_task()
    return server_prices

inventory_storages

inventory_storages(vendor)

List all storage types via boto3 calls.

Source code in sc_crawler/vendors/_aws.py

def inventory_storages(vendor):
    """List all storage types via `boto3` calls."""
    vendor.progress_tracker.start_task(
        name="Searching for storages", total=len(storage_manual_data)
    )

    # look up all volume types in us-east-1
    with ThreadPoolExecutor(max_workers=8) as executor:
        products = executor.map(
            _search_storage,
            storage_types,
            repeat(vendor),
            repeat("US East (N. Virginia)"),
        )
    products = list(chain.from_iterable(products))
    vendor.progress_tracker.hide_task()

    storages = []

    for product in products:
        attributes = product["product"]["attributes"]
        product_id = attributes["volumeApiName"]

        def get_attr(key: str) -> float:
            return extract_last_number(
                str(
                    attributes.get(
                        key,
                        storage_manual_data[product_id][key],
                    )
                )
            )

        storage_type = (
            StorageType.HDD if "HDD" in attributes["storageMedia"] else StorageType.SSD
        )
        storages.append(
            {
                "storage_id": product_id,
                "vendor_id": vendor.vendor_id,
                "name": attributes["volumeType"],
                "description": attributes["storageMedia"],
                "storage_type": storage_type,
                "max_iops": get_attr("maxIopsvolume"),
                "max_throughput": round(get_attr("maxThroughputvolume") * _MIB_TO_MB),
                "min_size": round(get_attr("minVolumeSize") * _GIB_TO_GB),
                "max_size": round(get_attr("maxVolumeSize") * 1024 * _GIB_TO_GB),
            }
        )

    return storages

inventory_storage_prices

inventory_storage_prices(vendor)

List all storage prices in all regions via boto3 calls.

Source code in sc_crawler/vendors/_aws.py

def inventory_storage_prices(vendor):
    """List all storage prices in all regions via `boto3` calls."""
    vendor.progress_tracker.start_task(
        name="Searching for storage_price(s)", total=len(storage_manual_data)
    )
    with ThreadPoolExecutor(max_workers=8) as executor:
        products = executor.map(
            _search_storage,
            storage_types,
            repeat(vendor),
        )
    products = list(chain.from_iterable(products))
    vendor.progress_tracker.hide_task()
    vendor.log(f"Found {len(products)} storage_price(s).")

    # lookup tables
    regions = scmodels_to_dict(vendor.regions, keys=["name", "aliases"])

    vendor.progress_tracker.start_task(
        name="Preprocessing storage_price(s)", total=len(products)
    )
    prices = []
    for product in products:
        try:
            attributes = product["product"]["attributes"]
            region = regions[attributes["location"]]
            price = _extract_ondemand_price(product["terms"])
            prices.append(
                {
                    "vendor_id": vendor.vendor_id,
                    "region_id": region.region_id,
                    "storage_id": attributes["volumeApiName"],
                    "unit": PriceUnit.GB_MONTH,
                    "price": float(price[0]),
                    "currency": price[1],
                }
            )
        except KeyError:
            continue
        finally:
            vendor.progress_tracker.advance_task()

    vendor.progress_tracker.hide_task()
    return prices

inventory_traffic_prices

inventory_traffic_prices(vendor)

List all inbound and outbound traffic prices in all regions via boto3 calls.

Source code in sc_crawler/vendors/_aws.py

def inventory_traffic_prices(vendor):
    """List all inbound and outbound traffic prices in all regions via `boto3` calls."""
    regions = scmodels_to_dict(vendor.regions, keys=["name", "aliases"])
    items = []
    for direction in list(TrafficDirection):
        loc_dir = "toLocation" if direction == TrafficDirection.IN else "fromLocation"
        vendor.progress_tracker.start_task(
            name=f"Searching for {direction.value} traffic_price(s)", total=None
        )
        products = _boto_get_products(
            service_code="AWSDataTransfer",
            filters={
                "transferType": "AWS " + direction.value.title(),
            },
        )
        vendor.log(f"Found {len(products)} {direction.value} traffic_price(s).")
        vendor.progress_tracker.update_task(
            description=f"Syncing {direction.value} traffic_price(s)",
            total=len(products),
        )
        for product in products:
            try:
                region = regions[product["product"]["attributes"][loc_dir]]
                prices = _extract_ondemand_prices(product["terms"], fix_1024=True)
                price = [PriceTier.model_validate(p).model_dump() for p in prices[0]]
                items.append(
                    {
                        "vendor_id": vendor.vendor_id,
                        "region_id": region.region_id,
                        "price": max([float(t["price"]) for t in prices[0]]),
                        "price_tiered": price,
                        "currency": prices[1],
                        "unit": PriceUnit.GB_MONTH,
                        "direction": direction,
                    }
                )
            except KeyError:
                continue
            finally:
                vendor.progress_tracker.advance_task()
        vendor.progress_tracker.hide_task()
    return items

inventory_ipv4_prices

inventory_ipv4_prices(vendor)

List IPV4 prices in all regions via boto3 calls.

Source code in sc_crawler/vendors/_aws.py

def inventory_ipv4_prices(vendor):
    """List IPV4 prices in all regions via `boto3` calls."""
    vendor.progress_tracker.start_task(name="Searching for ipv4_price(s)", total=None)
    products = _boto_get_products(
        service_code="AmazonVPC",
        filters={
            "group": "VPCPublicIPv4Address",
            "groupDescription": "Hourly charge for In-use Public IPv4 Addresses",
        },
    )
    vendor.log(f"Found {len(products)} ipv4_price(s).")
    vendor.progress_tracker.update_task(
        description="Syncing ipv4_price(s)", total=len(products)
    )
    # lookup tables
    regions = scmodels_to_dict(vendor.regions, keys=["name", "aliases"])
    items = []
    for product in products:
        try:
            region = regions[product["product"]["attributes"]["location"]]
        except KeyError as e:
            vendor.log("region not found: %s" % str(e), DEBUG)
            continue
        price = _extract_ondemand_price(product["terms"])
        items.append(
            {
                "vendor_id": vendor.vendor_id,
                "region_id": region.region_id,
                "price": float(price[0]),
                "currency": price[1],
                "unit": PriceUnit.HOUR,
            }
        )
        vendor.progress_tracker.advance_task()
    vendor.progress_tracker.hide_task()
    return items

inventory_databases

inventory_databases(vendor)

List all available AWS RDS PostgreSQL instance types via boto3 calls.

Source code in sc_crawler/vendors/_aws.py

def inventory_databases(vendor):
    """List all available AWS RDS PostgreSQL instance types via `boto3` calls."""
    vendor.progress_tracker.start_task(name="Searching for database(s)", total=None)
    prices_by_region, deployment_options_by_database = (
        _get_rds_instance_products_by_region()
    )
    vendor.progress_tracker.hide_task()

    regions = _active_region_ids(vendor)
    major_versions = _boto_describe_db_major_engine_versions_first(regions)
    options_by_database = _lookup_orderable_db_instance_options(
        vendor, prices_by_region
    )

    seen_database_ids = set()
    rows = []
    vendor.progress_tracker.start_task(
        name="Preprocessing database(s)", total=len(regions)
    )
    for region_id in regions:
        for database_id, product in prices_by_region.get(region_id, {}).items():
            if database_id in seen_database_ids:
                continue
            seen_database_ids.add(database_id)
            db_instance_options = options_by_database.get(database_id, [])
            status = Status.ACTIVE
            # Pricing keeps previous-gen SKUs (currentGeneration=No) after AWS stops
            # allowing new launches; add these classes with INACTIVE status.
            if not db_instance_options:
                status = Status.INACTIVE
            deployment_options = deployment_options_by_database.get(
                database_id, frozenset()
            )
            min_storage = [
                opt["MinStorageSize"]
                for opt in db_instance_options
                if opt.get("MinStorageSize") is not None
            ]
            max_storage = [
                opt["MaxStorageSize"]
                for opt in db_instance_options
                if opt.get("MaxStorageSize") is not None
            ]
            # https://docs.aws.amazon.com/AmazonRDS/latest/APIReference/API_OrderableDBInstanceOption.html
            # SupportsGlobalDatabases is Aurora Global Database only (not RDS postgres).
            multi_az = any(
                opt.get("MultiAZCapable") for opt in db_instance_options
            ) or ("Multi-AZ" in deployment_options)
            readable = "Multi-AZ (readable standbys)" in deployment_options
            ha = []
            ha_strategy = []
            if readable or multi_az:
                ha.append(DatabaseHaLevel.MULTI_ZONE)
            if "Single-AZ" in deployment_options:
                ha.append(DatabaseHaLevel.NONE)
            if not ha:
                raise ValueError(
                    f"Unknown deployment option for database {database_id}: {product}"
                )
            if readable:
                # https://docs.aws.amazon.com/AmazonRDS/latest/UserGuide/multi-az-db-clusters-concepts.html
                ha_strategy.append(DatabaseHaStrategy.READABLE_CLUSTER)
            if multi_az:
                # https://docs.aws.amazon.com/AmazonRDS/latest/UserGuide/Concepts.MultiAZSingleStandby.html
                ha_strategy.append(DatabaseHaStrategy.PASSIVE_STANDBY)
            if DatabaseHaLevel.NONE in ha:
                ha_strategy.append(DatabaseHaStrategy.NONE)
            storage_extra_autosize = any(
                opt.get("SupportsStorageAutoscaling") for opt in db_instance_options
            )
            # Orderable Min/MaxStorageSize are GiB
            storage_extra_min = (
                round(min(min_storage) * _GIB_TO_GB) if min_storage else None
            )
            storage_extra_max = (
                round(max(max_storage) * _GIB_TO_GB) if max_storage else None
            )
            disk_encryption = any(
                opt.get("SupportsStorageEncryption") for opt in db_instance_options
            )
            system_monitoring = any(
                opt.get("SupportsEnhancedMonitoring") for opt in db_instance_options
            )
            database_monitoring = any(
                opt.get("SupportsPerformanceInsights") for opt in db_instance_options
            )
            max_read_replicas = (
                15
                if any(opt.get("ReadReplicaCapable") for opt in db_instance_options)
                else 0
            )
            server_id = next(
                (
                    server.server_id
                    for server in vendor.servers
                    if server.api_reference == database_id.removeprefix("db.")
                ),
                None,
            )
            family = product["instanceFamily"]
            vcpus = product["vcpu"]
            memory_amount_gib = float(product["memory"].removesuffix(" GiB"))
            memory_amount_mib = int(memory_amount_gib * 1024)
            storage_size = _extract_rds_bundled_storage_size(product.get("storage"))
            description = (
                f"{family} ({vcpus} vCPU, {memory_amount_gib} GiB RAM, {storage_size} GB NVMe SSD)"
                if storage_size
                else f"{family} ({vcpus} vCPU, {memory_amount_gib} GiB RAM)"
            )
            rows.append(
                {
                    "vendor_id": vendor.vendor_id,
                    "database_id": database_id,
                    "name": database_id,
                    "api_reference": database_id,
                    # https://www.pulumi.com/registry/packages/aws/api-docs/rds/instance/
                    "api_reference_object": {"instance_class": database_id},
                    "display_name": database_id,
                    "engine": DatabaseEngine.POSTGRESQL,
                    "wire_protocol": DatabaseWireProtocol.POSTGRESQL,
                    "engine_versions": major_versions,
                    "family": product["instanceFamily"],
                    "description": description,
                    "server_id": server_id,
                    "vcpus": product["vcpu"],
                    "memory_amount": memory_amount_mib,
                    "storage_size": storage_size,
                    "ha": ha,
                    "ha_strategy": ha_strategy,
                    # https://docs.aws.amazon.com/AmazonRDS/latest/UserGuide/Overview.RDSSecurityGroups.html
                    # https://docs.aws.amazon.com/AmazonRDS/latest/UserGuide/USER_VPC.html
                    # https://docs.aws.amazon.com/AmazonRDS/latest/UserGuide/USER_VPC.pc.html
                    # https://docs.aws.amazon.com/AmazonRDS/latest/UserGuide/UsingWithRDS.IAMDBAuth.html
                    # https://docs.aws.amazon.com/AmazonRDS/latest/UserGuide/UsingWithRDS.SSL.html
                    # https://docs.aws.amazon.com/AmazonRDS/latest/UserGuide/Overview.Encryption.Keys.html
                    # https://docs.aws.amazon.com/AmazonRDS/latest/UserGuide/Appendix.PostgreSQL.CommonDBATasks.pgaudit.html
                    # Security groups (IP filtering), VPC placement, PrivateLink, IAM DB auth,
                    # rds.force_ssl, KMS CMK, pgaudit. Server certs only (no client-cert auth).
                    "security_features": [
                        DatabaseSecurityFeature.IP_FILTERING,
                        DatabaseSecurityFeature.PRIVATE_NETWORK,
                        DatabaseSecurityFeature.NETWORK_PEERING,
                        DatabaseSecurityFeature.IDENTITY_BASED_AUTH,
                        DatabaseSecurityFeature.ENFORCED_TLS,
                        DatabaseSecurityFeature.CUSTOMER_MANAGED_KEYS,
                        DatabaseSecurityFeature.AUDIT_LOGGING,
                    ],
                    "storage_extra_autosize": storage_extra_autosize,
                    "storage_extra_min": storage_extra_min,
                    "storage_extra_max": storage_extra_max,
                    "disk_encryption": disk_encryption,
                    "system_monitoring": system_monitoring,
                    "database_monitoring": database_monitoring,
                    # https://docs.aws.amazon.com/AmazonRDS/latest/UserGuide/USER_PostgreSQL.Replication.ReadReplicas.Configuration.html
                    # https://docs.aws.amazon.com/AmazonRDS/latest/UserGuide/CHAP_Limits.html
                    # Up to 15 read replicas per primary when the instance class supports replicas.
                    "max_read_replicas": max_read_replicas,
                    # https://docs.aws.amazon.com/AmazonRDS/latest/UserGuide/USER_UpgradeDBInstance.PostgreSQL.Minor.html
                    # Auto minor version upgrade is a supported instance option for RDS PostgreSQL.
                    "auto_upgrade_versions": True,
                    # https://docs.aws.amazon.com/AmazonRDS/latest/UserGuide/USER_WorkingWithAutomatedBackups.html
                    # Managed RDS includes automated backups; not per-class in orderable API.
                    "scheduled_backups": True,
                    # https://docs.aws.amazon.com/AmazonRDS/latest/UserGuide/USER_WorkingWithAutomatedBackups.BackupRetention.html
                    # Product max backup / PITR retention (days); not per SKU in orderable API.
                    "continuous_backups": 35,
                    # https://docs.aws.amazon.com/AmazonRDS/latest/UserGuide/USER_WorkingWithParamGroups.html
                    # DB parameter groups are supported for all RDS PostgreSQL instances.
                    "custom_config": True,
                    # https://docs.aws.amazon.com/AmazonRDS/latest/UserGuide/PostgreSQL.Concepts.General.FeatureSupport.Extensions.html
                    # Supported PostgreSQL extensions can be installed on the instance.
                    "custom_extensions": True,
                    # https://docs.aws.amazon.com/AmazonRDS/latest/UserGuide/rds-proxy.html
                    # RDS Proxy provides managed connection pooling for PostgreSQL.
                    "connection_pool": True,
                    # https://docs.aws.amazon.com/AmazonRDS/latest/UserGuide/devops-guru-for-rds.html
                    # DevOps Guru for RDS analyzes Performance Insights and recommends actions.
                    "autotuning_advice": True,
                    # https://docs.aws.amazon.com/AmazonRDS/latest/UserGuide/devops-guru-for-rds.html
                    # Recommendations require operator action; the service does not auto-apply fixes.
                    "autotuning_apply": False,
                    # https://aws.amazon.com/rds/sla/
                    # Multi-AZ SLO 99.95%; Single-AZ SLO 99.5% (credit tables).
                    "sla": 99.95 if multi_az or readable else 99.5,
                    "status": status,
                }
            )
        vendor.progress_tracker.advance_task()
    vendor.progress_tracker.hide_task()
    return rows

inventory_database_prices

inventory_database_prices(vendor)

List all on-demand AWS RDS PostgreSQL prices in all regions via boto3 calls.

Source code in sc_crawler/vendors/_aws.py

def inventory_database_prices(vendor):
    """List all on-demand AWS RDS PostgreSQL prices in all regions via `boto3` calls."""
    vendor.progress_tracker.start_task(
        name="Searching for database_price(s)", total=None
    )
    products = _boto_get_rds_products()
    vendor.progress_tracker.hide_task()

    region_ids = _active_region_ids(vendor)
    databases = {database.database_id for database in vendor.databases}
    items = []
    vendor.progress_tracker.start_task(
        name="Preprocessing database_price(s)", total=len(products)
    )
    for product in products:
        try:
            if product["product"].get("productFamily") != "Database Instance":
                continue
            attrs = product["product"]["attributes"]
            region_id = attrs["regionCode"]
            if region_id not in region_ids:
                continue
            if attrs.get("deploymentOption") == "Single-AZ":
                ha = DatabaseHaLevel.SINGLE_ZONE
                ha_strategy = DatabaseHaStrategy.NONE
            elif attrs.get("deploymentOption") == "Multi-AZ (readable standbys)":
                ha = DatabaseHaLevel.MULTI_ZONE
                ha_strategy = DatabaseHaStrategy.READABLE_CLUSTER
            elif attrs.get("deploymentOption") == "Multi-AZ":
                ha = DatabaseHaLevel.MULTI_ZONE
                ha_strategy = DatabaseHaStrategy.PASSIVE_STANDBY
            else:
                continue
            database_id = attrs.get("instanceType")
            if database_id not in databases:
                continue
            price = _extract_ondemand_price(product["terms"])
            items.append(
                {
                    "vendor_id": vendor.vendor_id,
                    "region_id": region_id,
                    "database_id": database_id,
                    "allocation": Allocation.ONDEMAND,
                    "ha": ha,
                    "ha_strategy": ha_strategy,
                    "unit": PriceUnit.HOUR,
                    "price": float(price[0]),
                    "price_upfront": 0,
                    "price_tiered": [],
                    "currency": price[1],
                }
            )
        except KeyError:
            continue
        finally:
            vendor.progress_tracker.advance_task()
    vendor.progress_tracker.hide_task()
    return items

inventory_database_storages

inventory_database_storages(vendor)

List all AWS RDS PostgreSQL storage types via boto3 calls.

Source code in sc_crawler/vendors/_aws.py

def inventory_database_storages(vendor):
    """List all AWS RDS PostgreSQL storage types via `boto3` calls."""
    vendor.progress_tracker.start_task(
        name="Searching for database_storage(s)", total=None
    )
    prices_by_region, _ = _get_rds_instance_products_by_region()
    vendor.progress_tracker.hide_task()

    options_by_database = _lookup_orderable_db_instance_options(
        vendor, prices_by_region
    )
    storage_bounds = _get_storage_bounds_from_orderable_options(options_by_database)

    items = []
    vendor.progress_tracker.start_task(
        name="Preprocessing database_storage(s)", total=len(storage_bounds)
    )
    for storage_id, bounds in storage_bounds.items():
        description = "HDD-backed" if storage_id == "standard" else "SSD-backed"
        with sentry_capture_or_raise(vendor=vendor):
            items.append(
                {
                    "vendor_id": vendor.vendor_id,
                    "database_storage_id": storage_id,
                    "name": _DATABASE_STORAGE_MAPPING[storage_id],
                    "description": description,
                    "scope": DatabaseStorageScope.DATA,
                    "min_size": bounds.get("min_size"),
                    "max_size": bounds.get("max_size"),
                    "max_iops": bounds.get("max_iops"),
                    "max_throughput": bounds.get("max_throughput"),
                }
            )
        vendor.progress_tracker.advance_task()
    vendor.progress_tracker.hide_task()
    return items

inventory_database_storage_prices

inventory_database_storage_prices(vendor)

List all AWS RDS PostgreSQL storage prices in all regions via boto3 calls.

Source code in sc_crawler/vendors/_aws.py

def inventory_database_storage_prices(vendor):
    """List all AWS RDS PostgreSQL storage prices in all regions via `boto3` calls."""
    vendor.progress_tracker.start_task(
        name="Searching for database_storage_price(s)", total=None
    )
    products = _boto_get_rds_products()
    vendor.progress_tracker.hide_task()

    region_ids = _active_region_ids(vendor)
    items = []
    database_storage_ids = [s.database_storage_id for s in vendor.database_storages]
    vendor.progress_tracker.start_task(
        name="Preprocessing database_storage_price(s)", total=len(products)
    )
    for product in products:
        try:
            if product["product"].get("productFamily") != "Database Storage":
                continue
            attrs = product["product"]["attributes"]
            region_id = attrs["regionCode"]
            if region_id not in region_ids:
                continue
            if attrs.get("deploymentOption") != "Single-AZ":
                continue
            storage_id = next(
                (
                    k
                    for k, v in _DATABASE_STORAGE_MAPPING.items()
                    if v == attrs.get("volumeType")
                ),
                None,
            )
            # Skip storage options that are not available in orderable API anymore
            if not storage_id or storage_id not in database_storage_ids:
                continue
            price = _extract_ondemand_price(product["terms"])
            items.append(
                {
                    "vendor_id": vendor.vendor_id,
                    "region_id": region_id,
                    "database_storage_id": storage_id,
                    "unit": PriceUnit.GB_MONTH,
                    "price": float(price[0]),
                    "price_upfront": 0,
                    "price_tiered": [],
                    "currency": price[1],
                }
            )
        except KeyError:
            continue
        finally:
            vendor.progress_tracker.advance_task()
    vendor.progress_tracker.hide_task()
    return items