_azure

sc_crawler.vendors._azure

Functions:

Name	Description
inventory_compliance_frameworks	Manual list of known compliance frameworks at Azure.
inventory_regions	List all regions via API call.
inventory_zones	List all availability zones.
inventory_servers	List all available instance types in all regions.
inventory_server_prices	List all known server ondemand prices in all regions using the Azure Retail Pricing API.
inventory_server_prices_spot	List all known server spot prices in all regions using the Azure Retail Pricing API.
inventory_storages	List all storage options via the Compute resource manager client.
inventory_storage_prices	Look up Storage prices via the Azure Retail Prices API.
inventory_traffic_prices	Look up Internet Egress/Ingress prices via the Azure Retail Prices API.
inventory_ipv4_prices	Look up Internet Egress/Ingress prices via the Azure Retail Prices API.
inventory_databases	List all available Azure Database for PostgreSQL Flexible Server types in all regions.
inventory_database_prices	List all known Azure Database for PostgreSQL Flexible Server on-demand prices in all regions using the Azure Retail Pricing API.
inventory_database_storages	List all Azure Database for PostgreSQL Flexible Server storage options via the capabilities API.
inventory_database_storage_prices	Look up Azure Database for PostgreSQL Flexible Server storage prices via the Azure Retail Prices API.

Attributes:

Name	Type	Description
SERVER_FEATURES		Map lowercase chars from the server name to features.
STORAGE_METER_MAPPING		Map Storage price meter names to the storage type ID and the disk size in GiB.
STORAGE_PRICE_UNIT_MAPPING	dict[str, float | None]	Storage capacity units → multiplier to convert the raw API price to $/GB/month.

SERVER_FEATURES module-attribute

SERVER_FEATURES = {'a': 'AMD processor', 'p': 'ARM processor', 'b': 'Block Storage performance', 'd': 'Local Disk', 'i': 'Isolated', 'l': 'Low Memory', 'm': 'Memory Intensive', 'n': 'Network Optimized', 't': 'Tiny Memory', 's': 'Premium Storage capable', 'r': 'RDMA capable', 'e': 'Memory Optimized', 'x': 'Unmatched Memory Capacity', 'o': 'o'}

Map lowercase chars from the server name to features.

STORAGE_METER_MAPPING module-attribute

STORAGE_METER_MAPPING = {'P1 LRS Disk': ('Premium_LRS', 4), 'P2 LRS Disk': ('Premium_LRS', 8), 'P3 LRS Disk': ('Premium_LRS', 16), 'P4 LRS Disk': ('Premium_LRS', 32), 'P6 LRS Disk': ('Premium_LRS', 64), 'P10 LRS Disk': ('Premium_LRS', 128), 'P15 LRS Disk': ('Premium_LRS', 256), 'P20 LRS Disk': ('Premium_LRS', 512), 'P30 LRS Disk': ('Premium_LRS', 1024), 'P40 LRS Disk': ('Premium_LRS', 2048), 'P50 LRS Disk': ('Premium_LRS', 4096), 'P60 LRS Disk': ('Premium_LRS', 8192), 'P70 LRS Disk': ('Premium_LRS', 16384), 'P80 LRS Disk': ('Premium_LRS', 32767), 'P1 ZRS Disk': ('Premium_ZRS', 4), 'P2 ZRS Disk': ('Premium_ZRS', 8), 'P3 ZRS Disk': ('Premium_ZRS', 16), 'P4 ZRS Disk': ('Premium_ZRS', 32), 'P6 ZRS Disk': ('Premium_ZRS', 64), 'P10 ZRS Disk': ('Premium_ZRS', 128), 'P15 ZRS Disk': ('Premium_ZRS', 256), 'P20 ZRS Disk': ('Premium_ZRS', 512), 'P30 ZRS Disk': ('Premium_ZRS', 1024), 'P40 ZRS Disk': ('Premium_ZRS', 2048), 'P50 ZRS Disk': ('Premium_ZRS', 4096), 'P60 ZRS Disk': ('Premium_ZRS', 8192), 'P70 ZRS Disk': ('Premium_ZRS', 16384), 'P80 ZRS Disk': ('Premium_ZRS', 32767), 'E1 LRS Disk': ('StandardSSD_LRS', 4), 'E2 LRS Disk': ('StandardSSD_LRS', 8), 'E3 LRS Disk': ('StandardSSD_LRS', 16), 'E4 LRS Disk': ('StandardSSD_LRS', 32), 'E6 LRS Disk': ('StandardSSD_LRS', 64), 'E10 LRS Disk': ('StandardSSD_LRS', 128), 'E15 LRS Disk': ('StandardSSD_LRS', 256), 'E20 LRS Disk': ('StandardSSD_LRS', 512), 'E30 LRS Disk': ('StandardSSD_LRS', 1024), 'E40 LRS Disk': ('StandardSSD_LRS', 2048), 'E50 LRS Disk': ('StandardSSD_LRS', 4096), 'E60 LRS Disk': ('StandardSSD_LRS', 8192), 'E70 LRS Disk': ('StandardSSD_LRS', 16384), 'E80 LRS Disk': ('StandardSSD_LRS', 32767), 'E1 ZRS Disk': ('StandardSSD_ZRS', 4), 'E2 ZRS Disk': ('StandardSSD_ZRS', 8), 'E3 ZRS Disk': ('StandardSSD_ZRS', 16), 'E4 ZRS Disk': ('StandardSSD_ZRS', 32), 'E6 ZRS Disk': ('StandardSSD_ZRS', 64), 'E10 ZRS Disk': ('StandardSSD_ZRS', 128), 'E15 ZRS Disk': ('StandardSSD_ZRS', 256), 'E20 ZRS Disk': ('StandardSSD_ZRS', 512), 'E30 ZRS Disk': ('StandardSSD_ZRS', 1024), 'E40 ZRS Disk': ('StandardSSD_ZRS', 2048), 'E50 ZRS Disk': ('StandardSSD_ZRS', 4096), 'E60 ZRS Disk': ('StandardSSD_ZRS', 8192), 'E70 ZRS Disk': ('StandardSSD_ZRS', 16384), 'E80 ZRS Disk': ('StandardSSD_ZRS', 32767), 'S4 LRS Disk': ('Standard_LRS', 32), 'S6 LRS Disk': ('Standard_LRS', 64), 'S10 LRS Disk': ('Standard_LRS', 128), 'S15 LRS Disk': ('Standard_LRS', 256), 'S20 LRS Disk': ('Standard_LRS', 512), 'S30 LRS Disk': ('Standard_LRS', 1024), 'S40 LRS Disk': ('Standard_LRS', 2048), 'S50 LRS Disk': ('Standard_LRS', 4096), 'S60 LRS Disk': ('Standard_LRS', 8192), 'S70 LRS Disk': ('Standard_LRS', 16384), 'S80 LRS Disk': ('Standard_LRS', 32767), 'Ultra LRS Provisioned Capacity': ('UltraSSD_LRS', 1), 'Premium LRS Provisioned Capacity': ('PremiumV2_LRS', 1)}

Map Storage price meter names to the storage type ID and the disk size in GiB.

Disk sizes sourced from the Azure Managed Disks pricing page: https://azure.microsoft.com/en-us/pricing/details/managed-disks/

Meter names sourced from the Azure Retail Prices API: https://learn.microsoft.com/en-us/rest/api/cost-management/retail-prices/azure-retail-prices

The disk size (second tuple element) is used to convert per-disk (1/Month) prices to $/GiB/month by dividing the retail price by this value. For provisioned-capacity meters (1 GiB/Hour) the divisor is 1 because the price is already expressed per GiB.

The following meter categories are intentionally not included:

• * Disk Mount — the extra monthly fee charged per VM when a disk is attached to more than one VM simultaneously (Azure Shared Disks feature). This is an access/ attach charge, not a storage-capacity charge.
• Confidential Compute Encryption * Provisioned Capacity — surcharge for customer-managed key (CMK) encryption on confidential VMs.
• LRS/ZRS Burst Enablement — monthly fee to permanently enable burst IOPS on Premium SSD disks.
• * Disk - Free / SAP Special * — zero-price promotional entries.
• Instant Access Snapshots — snapshot restore-access fee.
• Standard HDD ZRS is not offered as per-disk pricing (only as per-GiB/month).

STORAGE_PRICE_UNIT_MAPPING module-attribute

STORAGE_PRICE_UNIT_MAPPING = {'1/Month': None, '1 GiB/Month': 1 / _GIB_TO_GB, '1 GB/Month': 1.0, '1 GiB/Hour': _HOURS_PER_MONTH / _GIB_TO_GB, '1 GB/Hour': _HOURS_PER_MONTH}

Storage capacity units → multiplier to convert the raw API price to $/GB/month.

inventory_compliance_frameworks

inventory_compliance_frameworks(vendor)

Manual list of known compliance frameworks at Azure.

Data collected from https://learn.microsoft.com/en-us/azure/compliance/.

Source code in sc_crawler/vendors/_azure.py

def inventory_compliance_frameworks(vendor):
    """Manual list of known compliance frameworks at Azure.

    Data collected from <https://learn.microsoft.com/en-us/azure/compliance/>.
    """
    return map_compliance_frameworks_to_vendor(
        vendor.vendor_id, ["hipaa", "soc2t2", "iso27001"]
    )

inventory_regions

inventory_regions(vendor)

List all regions via API call.

Location (country and state) and founding year were collected manually from https://datacenters.microsoft.com/globe/explore/ and its underlying JSON at https://datacenters.microsoft.com/globe/data/geo/regions.json.

City and the energy source information was collected from the sustainability fact sheets referenced in the above page and JSON.

Coordinates were provided by the Microsoft API, which doesn't seem to be very reliable.

Source code in sc_crawler/vendors/_azure.py

def inventory_regions(vendor):
    """List all regions via API call.

    Location (country and state) and founding year
    were collected manually from
    <https://datacenters.microsoft.com/globe/explore/>
    and its underlying JSON at
    <https://datacenters.microsoft.com/globe/data/geo/regions.json>.

    City and the energy source information was collected from
    the sustainability fact sheets referenced in the above page and JSON.

    Coordinates were provided by the Microsoft API, which doesn't seem
    to be very reliable.
    """

    manual_datas = {
        # Canada
        "canadaeast": {
            "country_id": "CA",
            "state": "Quebec",
            "city": "Quebec City",
            "founding_year": 2016,
            # unknown as no sustainability fact sheet found
            "green_energy": False,
        },
        "canadacentral": {
            "country_id": "CA",
            "city": "Toronto",
            "founding_year": 2016,
            # unknown as no sustainability fact sheet found
            "green_energy": False,
        },
        # United States
        "centralus": {
            "country_id": "US",
            "state": "Iowa",
            "founding_year": 2014,
            "green_energy": True,
        },
        "centraluseuap": {
            "country_id": "US",
            "state": "Iowa",
            "green_energy": True,
        },
        "eastus": {
            "country_id": "US",
            "city": "Boydton",
            "state": "Virginia",
            # official site says 2014 with a dead link, but it was 2012 as per
            # https://web.archive.org/web/20120530115120/http:/blogs.msdn.com/b/windowsazure/archive/2012/04/05/announcing-new-datacenter-options-for-windows-azure.aspx
            "founding_year": 2012,
            "green_energy": False,
        },
        "eastusstg": {
            "country_id": "US",
            "state": "Virginia",
            "green_energy": False,
        },
        "eastus2": {
            "country_id": "US",
            "city": "Boydton",
            "state": "Virginia",
            # official site says 2012 with a dead link, but it was 2014 as per
            # https://azure.microsoft.com/en-us/updates/general-availability-microsoft-azure-us-central-and-us-east-2-regions/
            "founding_year": 2014,
            "green_energy": False,
        },
        "eastus2euap": {
            "country_id": "US",
            "state": "Virginia",
            "green_energy": False,
        },
        "northcentralus": {
            "country_id": "US",
            "city": "Chicago",
            "state": "Illinois",
            "founding_year": 2009,
            "green_energy": False,
        },
        "southcentralus": {
            "country_id": "US",
            "state": "Texas",
            "city": "San Antonio",
            "founding_year": 2008,
            "green_energy": True,
        },
        "southcentralusstg": {
            "country_id": "US",
            "state": "Texas",
            "city": "San Antonio",
        },
        "westcentralus": {
            "country_id": "US",
            "state": "Wyoming",
            "city": "Cheyenne",
            "founding_year": 2016,
            "green_energy": False,
        },
        "westus": {
            "country_id": "US",
            "state": "California",
            "founding_year": 2012,
            # unknown as no sustainability fact sheet found
            "green_energy": False,
        },
        "westus2": {
            "country_id": "US",
            "state": "Washington",
            "founding_year": 2007,
            "green_energy": False,
        },
        "westus3": {
            "country_id": "US",
            "state": "Arizona",
            "city": "Phoenix",
            "founding_year": 2021,
            "green_energy": False,
        },
        # Mexico
        "mexicocentral": {
            "country_id": "ZA",
            "state": "Querétaro",
            "founding_year": 2024,
            # unknown as no sustainability fact sheet found
            "green_energy": False,
        },
        # South America
        "brazilsouth": {
            "country_id": "BR",
            "state": "Campinas",
            "founding_year": 2014,
            "green_energy": False,
        },
        "brazilsoutheast": {
            "country_id": "US",
            "city": "Rio de Janeiro",
            # unknown as no sustainability fact sheet found
            "green_energy": False,
        },
        "chilecentral": {
            "country_id": "CL",
            "city": "Santiago",
            # coming soon
            "founding_year": 2025,
            # unknown as no sustainability fact sheet found
            "green_energy": False,
        },
        # not production region?
        # https://github.com/Azure/azure-dev/issues/2165#issuecomment-1542948509
        "brazilus": {
            "country_id": "BR",
        },
        # Asia Pacific
        "australiacentral": {
            "country_id": "AU",
            "city": "Canberra",
            "founding_year": 2018,
            # unknown as no sustainability fact sheet found
            "green_energy": False,
        },
        "australiacentral2": {
            "country_id": "AU",
            "city": "Canberra",
            "founding_year": 2018,
            # unknown as no sustainability fact sheet found
            "green_energy": False,
        },
        "australiaeast": {
            "country_id": "AU",
            "city": "Sydney",
            "state": "New South Wales",
            "founding_year": 2014,
            "green_energy": False,
        },
        "australiasoutheast": {
            "country_id": "AU",
            "city": "Melbourne",
            "state": "Victoria",
            "founding_year": 2014,
            "green_energy": False,
        },
        "eastasia": {
            "country_id": "HK",
            "founding_year": 2010,
            "green_energy": False,
        },
        "southeastasia": {
            "country_id": "SG",
            "city": "Singapore",
            "founding_year": 2010,
            "green_energy": False,
        },
        "japaneast": {
            "country_id": "JP",
            "city": "Tokyo",
            "founding_year": 2014,
            # unknown as no sustainability fact sheet found
            "green_energy": False,
        },
        "japanwest": {
            "country_id": "JP",
            "city": "Osaka",
            "founding_year": 2014,
            # unknown as no sustainability fact sheet found
            "green_energy": False,
        },
        "jioindiacentral": {
            "country_id": "IN",
            "city": "Nagpur",
            # unknown as no sustainability fact sheet found
            "green_energy": False,
        },
        "jioindiawest": {
            "country_id": "IN",
            "city": "Jamnagar",
            # unknown as no sustainability fact sheet found
            "green_energy": False,
        },
        "centralindia": {
            "country_id": "IN",
            "state": "Pune",
            "founding_year": 2015,
            # unknown as no sustainability fact sheet found
            "green_energy": False,
        },
        "southindia": {
            "country_id": "IN",
            "state": "Chennai",
            "founding_year": 2015,
            # unknown as no sustainability fact sheet found
            "green_energy": False,
        },
        "indiasouthcentral": {
            "country_id": "IN",
            "state": "Telangana",
            "city": "Hyderabad",
            # coming soon; targeted mid-2026 (Microsoft Source Asia, Dec 2025)
            "founding_year": 2026,
            # no region-specific sustainability fact sheet yet
            "green_energy": False,
        },
        "westindia": {
            "country_id": "IN",
            "state": "Mumbai",
            # unknown as no sustainability fact sheet found
            "green_energy": False,
        },
        "koreacentral": {
            "country_id": "KR",
            "city": "Seoul",
            "founding_year": 2017,
            # unknown as no sustainability fact sheet found
            "green_energy": False,
        },
        "koreasouth": {
            "country_id": "KR",
            "city": "Busan",
            # unknown as no sustainability fact sheet found
            "green_energy": False,
        },
        "newzealandnorth": {
            "country_id": "NZ",
            "city": "Auckland",
            "founding_year": 2024,
            "green_energy": False,
        },
        "indonesiacentral": {
            "country_id": "ID",
            "city": "Jakarta",
            # coming soon
            "founding_year": 2025,
            # unknown as no sustainability fact sheet found
            "green_energy": False,
        },
        "malaysiawest": {
            "country_id": "MY",
            "city": "Kuala Lumpur",
            # coming soon
            "founding_year": 2025,
            # unknown as no sustainability fact sheet found
            "green_energy": False,
        },
        # Europe
        "denmarkeast": {
            "country_id": "DK",
            "city": "Copenhagen",
            "founding_year": 2026,
            # https://news.microsoft.com/source/emea/features/accelerating-europes-digital-future-microsoft-announces-plans-for-a-new-datacenter-region-in-west-denmark
            "green_energy": True,
        },
        "francecentral": {
            "country_id": "FR",
            "city": "Paris",
            "founding_year": 2018,
            # unknown as no sustainability fact sheet found
            "green_energy": False,
        },
        "francesouth": {
            "country_id": "FR",
            "city": "Marseille",
            # unknown as no sustainability fact sheet found
            "green_energy": False,
        },
        "germanynorth": {
            "country_id": "DE",
            "city": "Berlin",
            # unknown as no sustainability fact sheet found
            "green_energy": False,
        },
        "germanywestcentral": {
            "country_id": "DE",
            "city": "Frankfurt",
            "founding_year": 2019,
            # unknown as no sustainability fact sheet found
            "green_energy": False,
        },
        "italynorth": {
            "country_id": "IT",
            "city": "Milan",
            "founding_year": 2023,
            "green_energy": False,
        },
        "northeurope": {
            "country_id": "IE",
            "city": "Dublin",
            "founding_year": 2009,
            "green_energy": False,
        },
        "norwayeast": {
            "country_id": "NO",
            "city": "Oslo",
            "founding_year": 2019,
            # unknown as no sustainability fact sheet found
            "green_energy": False,
        },
        "norwaywest": {
            "country_id": "NO",
        },
        "polandcentral": {
            "country_id": "PL",
            "city": "Warsaw",
            "founding_year": 2023,
            "green_energy": False,
        },
        "spaincentral": {
            "country_id": "ES",
            "city": "Madrid",
            "founding_year": 2024,
            "green_energy": False,
        },
        "swedencentral": {
            "country_id": "SE",
            "city": "Gävle and Sandviken",
            "founding_year": 2021,
            "green_energy": False,
        },
        "switzerlandnorth": {
            "country_id": "CH",
            "city": "Zürich",
            "founding_year": 2019,
            # unknown as no sustainability fact sheet found
            "green_energy": False,
        },
        "switzerlandwest": {
            "country_id": "CH",
            "city": "Geneva",
            # unknown as no sustainability fact sheet found
            "green_energy": False,
        },
        "uksouth": {
            "country_id": "GB",
            "city": "London",
            "founding_year": 2016,
            # unknown as no sustainability fact sheet found
            "green_energy": False,
        },
        "ukwest": {
            "country_id": "GB",
            "city": "Cardiff",
            "founding_year": 2017,
            # unknown as no sustainability fact sheet found
            "green_energy": False,
        },
        "westeurope": {
            "country_id": "NL",
            "founding_year": 2010,
            "green_energy": False,
        },
        "belgiumcentral": {
            "country_id": "BE",
            "founding_year": 2025,
            # unknown as no sustainability fact sheet found
            "green_energy": False,
        },
        # Middle East
        "israelcentral": {
            "country_id": "IL",
            "founding_year": 2023,
            # unknown as no sustainability fact sheet found
            "green_energy": False,
        },
        "qatarcentral": {
            "country_id": "QA",
            "city": "Doha",
            "founding_year": 2022,
            # unknown as no sustainability fact sheet found
            "green_energy": False,
        },
        "uaecentral": {
            "country_id": "AE",
            "city": "Abu Dhabi",
        },
        "uaenorth": {
            "country_id": "AE",
            "city": "Dubai",
            "founding_year": 2019,
            # unknown as no sustainability fact sheet found
            "green_energy": False,
        },
        "austriaeast": {
            "country_id": "AT",
            "city": "Vienna",
            "founding_year": 2025,
            # unknown as no sustainability fact sheet found
            "green_energy": False,
        },
        # Africa
        "southafricanorth": {
            "country_id": "ZA",
            "city": "Johannesburg",
            "founding_year": 2019,
            # unknown as no sustainability fact sheet found
            "green_energy": False,
        },
        "southafricawest": {
            "country_id": "ZA",
            "city": "Cape Town",
            # unknown as no sustainability fact sheet found
            "green_energy": False,
        },
        # China TODO enable
    }

    items = []
    for region in _regions():
        with sentry_capture_or_raise(vendor=vendor):
            # TODO drop this once the metadata field doesn't show up randomly anymore
            # as the non-metadata responses do not seem to have these logical regions anymore
            metadata = region.get("metadata", {})
            region_type = metadata.get("regionType") or metadata.get(
                "region_type", "Physical"
            )
            if region_type != "Physical":
                continue
            # no idea what are these
            if region["name"].endswith("stg"):
                continue
            if region["name"].endswith("stage"):
                continue
            # not production region?
            # https://github.com/Azure/azure-dev/issues/2165#issuecomment-1542948509
            if region["name"] == "brazilus":
                continue
            # exclude for now as this new region is popping up and being removed
            # from their API response randomly, so messing with git history
            if region["name"] == "newzealandnorth":
                continue
            manual_data = manual_datas.get(region["name"])
            if not manual_data:
                raise KeyError(f"No manual data found for {region['name']}.")
            # azure-mgmt-resource 25+ as_dict() uses camelCase keys (displayName)
            region_name = next(
                (region[k] for k in ("display_name", "displayName") if region.get(k)),
                region["name"],
            )
            display_name = region.get("displayName") or region.get("display_name")
            items.append(
                {
                    "vendor_id": vendor.vendor_id,
                    "region_id": region["name"],
                    "name": region_name,
                    "api_reference": region["name"],
                    "display_name": (
                        region_name + " (" + manual_data["country_id"] + ")"
                    ),
                    "aliases": [display_name] if display_name else [],
                    "country_id": manual_data["country_id"],
                    "state": manual_data.get("state"),
                    "city": manual_data.get("city"),
                    "address_line": None,
                    "zip_code": None,
                    # sometimes the API passes "metadata" and the lat/long nested, sometimes it's not
                    # TODO revisit after a few days passed since this change:
                    # https://github.com/Azure/azure-sdk-for-python/blob/azure-mgmt-resource_25.0.0/sdk/resources/azure-mgmt-resource/CHANGELOG.md#2500-2026-02-04
                    "lat": region.get("metadata", {}).get(
                        "latitude", region.get("latitude")
                    ),
                    "lon": region.get("metadata", {}).get(
                        "longitude", region.get("longitude")
                    ),
                    "founding_year": manual_data.get("founding_year"),
                    "green_energy": manual_data.get("green_energy"),
                }
            )
    return items

inventory_zones

inventory_zones(vendor)

List all availability zones.

API call to list existing availability zones ("1", "2", and "3") for each region, and creating a dummy "0" zone for the regions without availability zones.

Source code in sc_crawler/vendors/_azure.py

def inventory_zones(vendor):
    """List all availability zones.

    API call to list existing availability zones ("1", "2", and "3")
    for each region, and creating a dummy "0" zone for the regions
    without availability zones.
    """
    items = []
    resources = _resources("Microsoft.Compute")
    locations = next(
        (
            i
            for i in resources
            if i.get("resource_type") == "virtualMachines"
            or i.get("resourceType") == "virtualMachines"
        ),
        {},
    )
    zone_mappings = locations.get("zone_mappings", locations.get("zoneMappings", []))
    locations = {item["location"]: item["zones"] for item in zone_mappings}
    for region in vendor.regions:
        # default to zone with 0 ID if there are no real availability zones
        region_zones = locations.get(region.name, ["0"])
        for zone in region_zones:
            items.append(
                {
                    "vendor_id": vendor.vendor_id,
                    "region_id": region.region_id,
                    "zone_id": zone,
                    "name": zone,
                    "api_reference": zone,
                    "display_name": region.region_id + "-" + zone,
                }
            )
    return items

inventory_servers

inventory_servers(vendor)

List all available instance types in all regions.

Source code in sc_crawler/vendors/_azure.py

def inventory_servers(vendor):
    """List all available instance types in all regions."""
    servers = _servers()
    for i in range(len(servers) - 1, -1, -1):
        name = servers[i].get("name")
        # drop Basic servers as to be deprecated by Aug 2024
        if name.startswith("Basic"):
            vendor.log(f"Excluding deprecated: {name}")
            servers.pop(i)
        # servers that are likely to be not available, with zero pricing
        if name.endswith("Promo"):
            vendor.log(f"Excluding nonsense pricing: {name}")
            servers.pop(i)
        # servers probably not intended for our eyes
        if "Internal" in name:
            vendor.log(f"Excluding internal server: {name}")
            servers.pop(i)
        # servers randomly switching between active/inactive status
        # TODO review from time to time
        if name in ["Standard_M896ixds_32_v3", "Standard_M64-32bds_1_v3"]:
            vendor.log(f"Excluding server with questionable availability: {name}")
            servers.pop(i)
    servers = preprocess_servers(servers, vendor, _standardize_server)
    return servers

inventory_server_prices

inventory_server_prices(vendor)

List all known server ondemand prices in all regions using the Azure Retail Pricing API.

More information: https://learn.microsoft.com/en-us/rest/api/cost-management/retail-prices/azure-retail-prices.

Source code in sc_crawler/vendors/_azure.py

def inventory_server_prices(vendor):
    """List all known server ondemand prices in all regions using the Azure Retail Pricing API.

    More information: <https://learn.microsoft.com/en-us/rest/api/cost-management/retail-prices/azure-retail-prices>.
    """
    return _inventory_server_prices(vendor, Allocation.ONDEMAND)

inventory_server_prices_spot

inventory_server_prices_spot(vendor)

List all known server spot prices in all regions using the Azure Retail Pricing API.

See details at inventory_server_prices.

Source code in sc_crawler/vendors/_azure.py

def inventory_server_prices_spot(vendor):
    """List all known server spot prices in all regions using the Azure Retail Pricing API.

    See details at [inventory_server_prices][sc_crawler.vendors._azure.inventory_server_prices].
    """
    return _inventory_server_prices(vendor, Allocation.SPOT)

inventory_storages

inventory_storages(vendor)

List all storage options via the Compute resource manager client.

For more information, see https://learn.microsoft.com/en-us/azure/virtual-machines/disks-types.

Source code in sc_crawler/vendors/_azure.py

def inventory_storages(vendor):
    """List all storage options via the Compute resource manager client.

    For more information, see <https://learn.microsoft.com/en-us/azure/virtual-machines/disks-types>.
    """
    vendor.progress_tracker.start_task(
        name="Fetching list of compute resources", total=None
    )

    disks = []
    for resource in _compute_resources():
        if resource["resourceType"] == "disks":
            disks.append(resource)

    disks = list({d["name"]: d for d in disks}.values())
    vendor.progress_tracker.hide_task()

    items = []
    for disk in disks:

        def _search(values):
            return list_search(disk["capabilities"], "name", values)["value"]

        storage_type = (
            StorageType.HDD
            if "Standard" in disk["name"] and "SSD" not in disk["name"]
            else StorageType.SSD
        )
        redundancy_type = (
            "Locally Redundant Storage"
            if "LRS" in disk["name"]
            else "Zone-Redundant Storage"
        )
        description = f"{disk['tier']} tier {storage_type.name} ({redundancy_type})"
        storage_sizes = [
            storage[1]
            for storage in STORAGE_METER_MAPPING.values()
            if storage[0] == disk["name"]
        ]
        min_size = (
            min(storage_sizes) if len(storage_sizes) > 1 else _search("MinSizeGiB")
        )
        max_size = (
            max(storage_sizes) if len(storage_sizes) > 1 else _search("MaxSizeGiB")
        )

        items.append(
            {
                "storage_id": disk["name"],
                "vendor_id": vendor.vendor_id,
                "name": disk["name"],
                "description": description,
                "storage_type": storage_type,
                "max_iops": _search(["MaxIOpsReadWrite", "MaxIOps"]),
                "max_throughput": _search(
                    ["MaxBandwidthMBpsReadWrite", "MaxBandwidthMBps"]
                ),
                "min_size": round(int(min_size) * _GIB_TO_GB),
                "max_size": round(int(max_size) * _GIB_TO_GB),
            }
        )
    return items

inventory_storage_prices

inventory_storage_prices(vendor)

Look up Storage prices via the Azure Retail Prices API.

For more information, see https://learn.microsoft.com/en-us/rest/api/cost-management/retail-prices/azure-retail-prices.

Source code in sc_crawler/vendors/_azure.py

def inventory_storage_prices(vendor):
    """Look up Storage prices via the Azure Retail Prices API.

    For more information, see <https://learn.microsoft.com/en-us/rest/api/cost-management/retail-prices/azure-retail-prices>.
    """
    vendor.progress_tracker.start_task(
        name="Fetching list of storage resources", total=None
    )
    retail_prices = []
    with sentry_capture_or_raise(vendor=vendor):
        retail_prices = _prices("$filter=serviceName eq 'Storage'")
    vendor.progress_tracker.hide_task()

    if not retail_prices:
        return []

    regions = scmodels_to_dict(vendor.regions, keys=["region_id"])
    storages = scmodels_to_dict(vendor.storages, keys=["storage_id"])

    prices = defaultdict(list)
    for p in retail_prices:
        mapping = STORAGE_METER_MAPPING.get(p["meterName"])
        if not (
            mapping
            and mapping[0] in storages.keys()
            and p["armRegionName"] in regions.keys()
        ):
            continue

        unit = p.get("unitOfMeasure")
        if unit == "1/Month":
            # mapping[1] is disk size in GiB, convert to $/GB/month.
            price = p["retailPrice"] / mapping[1] / _GIB_TO_GB
            prices[
                (
                    vendor.vendor_id,
                    p["armRegionName"],
                    mapping[0],
                    PriceUnit.GB_MONTH,
                    p["currencyCode"],
                )
            ].append(
                PriceTier(
                    lower=0,
                    upper=round(mapping[1] * _GIB_TO_GB, 4),
                    price=round(price, 4),
                )
            )
        else:
            multiplier = STORAGE_PRICE_UNIT_MAPPING.get(unit)
            if multiplier is None:
                continue  # not a storage capacity unit
            price = p["retailPrice"] * multiplier
            prices[
                (
                    vendor.vendor_id,
                    p["armRegionName"],
                    mapping[0],
                    PriceUnit.GB_MONTH,
                    p["currencyCode"],
                )
            ].append(PriceTier(lower=0, upper=float("inf"), price=round(price, 4)))

    items = []
    for key, price_tiered in prices.items():
        vendor_id, region_id, storage_id, unit, currency = key
        if len(price_tiered) == 1:
            price = price_tiered[0].price
            price_tiered = []
        else:
            price_tiered.sort(key=lambda x: x.upper)
            lower = 0
            for price_tier in price_tiered:
                price_tier.lower = lower
                lower = price_tier.upper
            price = price_tiered[0].price  # use the price of the lowest tier

        items.append(
            {
                "vendor_id": vendor_id,
                "region_id": region_id,
                "storage_id": storage_id,
                "unit": unit,
                "price": price,
                "price_tiered": price_tiered,
                "currency": currency,
            }
        )

    return items

inventory_traffic_prices

inventory_traffic_prices(vendor)

Look up Internet Egress/Ingress prices via the Azure Retail Prices API.

For more information, see https://learn.microsoft.com/en-us/rest/api/cost-management/retail-prices/azure-retail-prices.

Source code in sc_crawler/vendors/_azure.py

def inventory_traffic_prices(vendor):
    """Look up Internet Egress/Ingress prices via the Azure Retail Prices API.

    For more information, see <https://learn.microsoft.com/en-us/rest/api/cost-management/retail-prices/azure-retail-prices>.
    """

    def get_tiers(prices: List[dict]) -> List[dict]:
        def prep_tiers(d: dict) -> dict:
            return {
                "lower": d.get("tierMinimumUnits", 0),
                "price": d["retailPrice"],
            }

        tiers = [prep_tiers(p) for p in prices]
        tiers.sort(key=lambda x: x.get("lower"))
        for i in range(len(tiers)):
            if i == len(tiers) - 1:
                tiers[i]["upper"] = "Infinity"
            else:
                tiers[i]["upper"] = tiers[i + 1]["lower"]
        return tiers

    def by_region(prices: List[dict], region: str) -> List[dict]:
        return [p for p in prices if p["armRegionName"] == region]

    vendor.progress_tracker.start_task(
        name="Fetching list of traffic prices", total=None
    )
    inbound_prices = []
    outbound_prices = []
    with sentry_capture_or_raise(vendor=vendor):
        inbound_prices = _prices(
            "$filter=serviceFamily eq 'Networking' and meterName eq 'Standard Data Transfer In'"
        )
        outbound_prices = _prices(
            "$filter=serviceFamily eq 'Networking' and "
            "meterName eq 'Standard Data Transfer Out' and "
            "productName eq 'Bandwidth - Routing Preference: Internet'"
        )
    vendor.progress_tracker.hide_task()

    if not inbound_prices and not outbound_prices:
        return []

    items = []
    regions = scmodels_to_dict(vendor.regions, keys=["api_reference"])
    for region in regions.values():
        for direction in ["inbound", "outbound"]:
            prices = inbound_prices if direction == "inbound" else outbound_prices
            tiers = get_tiers(by_region(prices, region.api_reference))
            if tiers:
                items.append(
                    {
                        "vendor_id": vendor.vendor_id,
                        "region_id": region.region_id,
                        "price": max([float(t["price"]) for t in tiers]),
                        "price_tiered": tiers,
                        "currency": prices[0].get("currencyCode", "USD"),
                        "unit": PriceUnit.GB_MONTH,
                        "direction": (
                            TrafficDirection.IN
                            if direction == "inbound"
                            else TrafficDirection.OUT
                        ),
                    }
                )
    return items

inventory_ipv4_prices

inventory_ipv4_prices(vendor)

Look up Internet Egress/Ingress prices via the Azure Retail Prices API.

For more information, see https://learn.microsoft.com/en-us/rest/api/cost-management/retail-prices/azure-retail-prices.

Source code in sc_crawler/vendors/_azure.py

def inventory_ipv4_prices(vendor):
    """Look up Internet Egress/Ingress prices via the Azure Retail Prices API.

    For more information, see <https://learn.microsoft.com/en-us/rest/api/cost-management/retail-prices/azure-retail-prices>.
    """

    vendor.progress_tracker.start_task(
        name="Fetching list of traffic prices", total=None
    )
    prices = []
    with sentry_capture_or_raise(vendor=vendor):
        prices = _prices(
            "$filter=serviceFamily eq 'Networking' and "
            "meterName eq 'Basic IPv4 Dynamic Public IP' and "
            "type eq 'Consumption'"
        )
    vendor.progress_tracker.hide_task()

    if not prices:
        return []

    items = []
    regions = scmodels_to_dict(vendor.regions, keys=["api_reference"])
    for region in regions.values():
        price = list_search(prices, "armRegionName", region.api_reference)
        if price:
            items.append(
                {
                    "vendor_id": vendor.vendor_id,
                    "region_id": region.region_id,
                    "price": float(price["retailPrice"]),
                    "currency": price.get("currencyCode", "USD"),
                    "unit": PriceUnit.HOUR,
                }
            )
    return items

inventory_databases

inventory_databases(vendor)

List all available Azure Database for PostgreSQL Flexible Server types in all regions.

Source code in sc_crawler/vendors/_azure.py

def inventory_databases(vendor):
    """List all available Azure Database for PostgreSQL Flexible Server types in all regions."""
    merged: dict[str, dict] = {}
    regions = _pg_database_regions(vendor)
    vendor.progress_tracker.start_task(
        name="Scanning region(s) for database(s)", total=len(regions)
    )
    for region in regions:
        location = region.api_reference
        with sentry_capture_or_raise(vendor=vendor):
            for capability in _pg_capabilities(location):
                engine_versions = _pg_engine_versions(capability)
                storage_extra_autosize = None
                storage_auto_growth = getattr(
                    capability, "storage_auto_growth_supported", None
                )
                if storage_auto_growth == "Enabled":
                    storage_extra_autosize = True
                elif storage_auto_growth == "Disabled":
                    storage_extra_autosize = False
                # IndexTuning → advice; AdaptiveAutoVacuumAutoApply → apply (vacuum GUCs).
                # https://learn.microsoft.com/en-us/azure/postgresql/monitor/concepts-autonomous-tuning
                # https://learn.microsoft.com/en-us/azure/postgresql/monitor/concepts-adaptive-autovacuum
                autotuning_advice = False
                autotuning_apply = False
                for feature in getattr(capability, "supported_features", None) or []:
                    feature_name = getattr(feature, "name", None)
                    enabled = getattr(feature, "status", None) == "Enabled"
                    if (
                        feature_name == "StorageAutoGrowth"
                        and storage_extra_autosize is None
                    ):
                        storage_extra_autosize = enabled
                    elif feature_name == "IndexTuning":
                        autotuning_advice = enabled
                    elif feature_name == "AdaptiveAutoVacuumAutoApply":
                        autotuning_apply = enabled
                    # Maybe this the indicator for Writes tuning, but it's not documented.
                    # https://learn.microsoft.com/en-us/azure/postgresql/monitor/concepts-intelligent-tuning
                    # elif feature_name == "ConfigTuning":
                    #     autotuning_apply = enabled
                for edition in (
                    getattr(capability, "supported_server_editions", None) or []
                ):
                    sizes_gb = []
                    for storage_edition in (
                        getattr(edition, "supported_storage_editions", None) or []
                    ):
                        for storage_mb in (
                            getattr(storage_edition, "supported_storage_mb", None) or []
                        ):
                            size_mb = getattr(storage_mb, "storage_size_mb", None)
                            max_mb = getattr(
                                storage_mb, "maximum_storage_size_mb", None
                            )
                            # Azure storage_size_mb values are MiB
                            if size_mb is not None:
                                sizes_gb.append(round(int(size_mb) / 1024 * _GIB_TO_GB))
                            if max_mb is not None:
                                sizes_gb.append(round(int(max_mb) / 1024 * _GIB_TO_GB))
                    storage_extra_min = min(sizes_gb) if sizes_gb else None
                    storage_extra_max = max(sizes_gb) if sizes_gb else None
                    for sku in getattr(edition, "supported_server_skus", None) or []:
                        database_id = sku.name
                        server_id = next(
                            (
                                server.server_id
                                for server in vendor.servers
                                if server.api_reference == database_id
                            ),
                            None,
                        )
                        vcpus = int(sku.v_cores) if sku.v_cores else None
                        memory_amount = (
                            int(sku.supported_memory_per_vcore_mb * sku.v_cores)
                            if sku.v_cores and sku.supported_memory_per_vcore_mb
                            else None
                        )
                        spec_parts = []
                        if vcpus is not None:
                            spec_parts.append(
                                f"{vcpus} vCPU{'s' if vcpus != 1 else ''}"
                            )
                        if memory_amount is not None:
                            spec_parts.append(f"{memory_amount // 1024} GB RAM")
                        description = f"PostgreSQL {edition.name}"
                        if spec_parts:
                            description = f"{description} ({', '.join(spec_parts)})"
                        # https://learn.microsoft.com/en-us/azure/postgresql/high-availability/concepts-high-availability
                        # Flexible Server HA is same-region only (zone-redundant or zonal sync standby) and optional.
                        # Burstable does not support HA (API may still list modes - prefer docs).
                        # Cross-region geo read replicas are DR / read scale, not HA - see max_read_replicas.
                        ha_modes = {
                            (mode.value if hasattr(mode, "value") else str(mode))
                            for mode in (getattr(sku, "supported_ha_mode", None) or [])
                        }
                        ha: list[DatabaseHaLevel] = []
                        ha_strategy: list[DatabaseHaStrategy] = []
                        if edition.name != "Burstable":
                            if "ZoneRedundant" in ha_modes:
                                ha.append(DatabaseHaLevel.MULTI_ZONE)
                            if "SameZone" in ha_modes:
                                ha.append(DatabaseHaLevel.SINGLE_ZONE)
                            if ha:
                                ha_strategy.append(DatabaseHaStrategy.PASSIVE_STANDBY)
                        # Non-HA deployment is always available
                        ha.append(DatabaseHaLevel.NONE)
                        ha_strategy.append(DatabaseHaStrategy.NONE)
                        if earlier_ha := merged.get(database_id, {}).get("ha", []):
                            if DatabaseHaLevel.MULTI_ZONE in earlier_ha:
                                ha = earlier_ha
                        # https://learn.microsoft.com/en-us/azure/reliability/reliability-database-postgresql
                        # Zone-redundant HA 99.99%; zonal HA 99.95%; no HA 99.9%.
                        if DatabaseHaLevel.MULTI_ZONE in ha:
                            sla = 99.99
                        elif DatabaseHaLevel.SINGLE_ZONE in ha:
                            sla = 99.95
                        else:
                            sla = 99.9
                        merged[database_id] = {
                            "vendor_id": vendor.vendor_id,
                            "database_id": database_id,
                            "name": database_id.removeprefix("Standard_"),
                            "api_reference": database_id,
                            # https://www.pulumi.com/registry/packages/azure/api-docs/postgresql/flexibleserver/
                            # Pulumi sku_name is tier+name (e.g. B_Standard_B1ms, GP_Standard_D2s_v3).
                            "api_reference_object": {
                                "sku_name": (
                                    f"{_PG_SKU_NAME_PREFIX[edition.name]}_{database_id}"
                                    if edition.name in _PG_SKU_NAME_PREFIX
                                    else database_id
                                )
                            },
                            "display_name": database_id.removeprefix("Standard_"),
                            "description": description,
                            "server_id": server_id,
                            "engine": DatabaseEngine.POSTGRESQL,
                            "wire_protocol": DatabaseWireProtocol.POSTGRESQL,
                            "engine_versions": engine_versions,
                            "family": edition.name,
                            "vcpus": vcpus,
                            "memory_amount": memory_amount,
                            "storage_size": None,
                            "storage_extra_autosize": storage_extra_autosize,
                            # https://learn.microsoft.com/en-us/azure/postgresql/backup-restore/concepts-backup-restore
                            # Automated backups are always enabled for Flexible Server.
                            "scheduled_backups": True,
                            "autotuning_advice": autotuning_advice,
                            "autotuning_apply": autotuning_apply,
                            # https://learn.microsoft.com/en-us/azure/postgresql/security/security-overview#data-protection
                            "disk_encryption": True,
                            # https://learn.microsoft.com/en-us/azure/postgresql/configure-maintain/concepts-major-version-upgrade
                            # Minor releases are applied automatically during maintenance.
                            "auto_upgrade_versions": True,
                            # https://learn.microsoft.com/en-us/azure/postgresql/parameters/concepts-parameters
                            "custom_config": True,
                            # https://learn.microsoft.com/en-us/azure/postgresql/extensions/concepts-extensions-considerations
                            "custom_extensions": True,
                            "storage_extra_min": storage_extra_min,
                            "storage_extra_max": storage_extra_max,
                            "ha": ha,
                            "ha_strategy": ha_strategy,
                            # https://learn.microsoft.com/en-us/azure/postgresql/network/concepts-networking-public
                            # https://learn.microsoft.com/en-us/azure/postgresql/network/concepts-networking-private
                            # https://learn.microsoft.com/en-us/azure/postgresql/network/concepts-networking-private-link
                            # https://learn.microsoft.com/en-us/azure/postgresql/flexible-server/how-to-configure-sign-in-azure-ad-authentication
                            # https://learn.microsoft.com/en-us/azure/postgresql/flexible-server/how-to-connect-tls-ssl
                            # https://learn.microsoft.com/en-us/azure/postgresql/flexible-server/concepts-data-encryption
                            # https://learn.microsoft.com/en-us/azure/postgresql/flexible-server/concepts-audit
                            # Firewall rules, VNet injection, Private Link, Entra ID auth,
                            # require_secure_transport, Key Vault CMK, pgaudit. Server certs only.
                            "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,
                            ],
                            # https://learn.microsoft.com/en-us/azure/postgresql/read-replica/concepts-read-replicas
                            # Up to 5 replicas per primary; Burstable tier is not supported.
                            "max_read_replicas": (
                                0 if edition.name == "Burstable" else 5
                            ),
                            # https://learn.microsoft.com/en-us/azure/postgresql/connectivity/concepts-pgbouncer
                            # Built-in PgBouncer connection pooling.
                            "connection_pool": True,
                            # https://learn.microsoft.com/en-us/azure/postgresql/monitor/concepts-monitoring
                            # Host-level metrics via Azure Monitor.
                            "system_monitoring": True,
                            # https://learn.microsoft.com/en-us/azure/postgresql/monitor/concepts-query-performance-insight
                            # Query Performance Insight / Query Store for query-level analysis.
                            "database_monitoring": True,
                            # https://learn.microsoft.com/en-us/azure/postgresql/backup-restore/concepts-backup-restore
                            # Product max PITR retention (days); default 7, up to 35; not in capabilities API.
                            "continuous_backups": 35,
                            "sla": sla,
                        }
        vendor.progress_tracker.advance_task()
    vendor.progress_tracker.hide_task()
    return list(merged.values())

inventory_database_prices

inventory_database_prices(vendor)

List all known Azure Database for PostgreSQL Flexible Server on-demand prices in all regions using the Azure Retail Pricing API.

More information: https://learn.microsoft.com/en-us/rest/api/cost-management/retail-prices/azure-retail-prices.

Source code in sc_crawler/vendors/_azure.py

def inventory_database_prices(vendor):
    """List all known Azure Database for PostgreSQL Flexible Server on-demand prices in all regions using the Azure Retail Pricing API.

    More information: <https://learn.microsoft.com/en-us/rest/api/cost-management/retail-prices/azure-retail-prices>.
    """
    items = []
    seen: set[tuple[str, str, DatabaseHaLevel, DatabaseHaStrategy]] = set()
    regions = _pg_database_regions(vendor)
    vendor.progress_tracker.start_task(
        name="Scanning region(s) for database_price(s)", total=len(regions)
    )
    for region in regions:
        location = region.api_reference
        with sentry_capture_or_raise(vendor=vendor):
            prices_by_arm: dict[str, list[dict]] = {}
            for item in _pg_retail_prices(location):
                product = (item.get("productName") or "").lower()
                arm = item.get("armSkuName") or ""
                meter = item.get("meterName") or ""
                if not (
                    "compute" in product
                    or arm.startswith("Standard_")
                    or arm.upper() in _PG_FLAT_COMPUTE_SKUS
                    or meter.upper() in _PG_FLAT_COMPUTE_SKUS
                ):
                    continue
                if arm:
                    prices_by_arm.setdefault(arm, []).append(item)

            for capability in _pg_capabilities(location):
                for edition in (
                    getattr(capability, "supported_server_editions", None) or []
                ):
                    edition_name = edition.name
                    for sku in getattr(edition, "supported_server_skus", None) or []:
                        database_id = sku.name
                        if not database_id:
                            continue
                        price_item = _pg_lookup_retail_price(
                            database_id=database_id,
                            edition_name=edition_name,
                            prices_by_arm=prices_by_arm,
                        )
                        if price_item is None:
                            continue
                        vcpus = int(sku.v_cores) if sku.v_cores else None
                        base_price = _pg_hourly_compute_price(
                            price_item, vcpus, database_id=database_id
                        )
                        currency = price_item.get("currencyCode", "USD")
                        # https://azure.microsoft.com/pricing/details/postgresql/flexible-server/
                        # HA bills primary + standby at the same rate (exactly 2x); SameZone
                        # and ZoneRedundant cost the same. No separate retail HA meters.
                        ha_modes = {
                            (mode.value if hasattr(mode, "value") else str(mode))
                            for mode in (getattr(sku, "supported_ha_mode", None) or [])
                        }
                        price_rows: list[
                            tuple[DatabaseHaLevel, DatabaseHaStrategy, float]
                        ] = [
                            (DatabaseHaLevel.NONE, DatabaseHaStrategy.NONE, base_price)
                        ]
                        if edition_name != "Burstable":
                            if "ZoneRedundant" in ha_modes:
                                price_rows.append(
                                    (
                                        DatabaseHaLevel.MULTI_ZONE,
                                        DatabaseHaStrategy.PASSIVE_STANDBY,
                                        base_price * 2,
                                    )
                                )
                            if "SameZone" in ha_modes:
                                price_rows.append(
                                    (
                                        DatabaseHaLevel.SINGLE_ZONE,
                                        DatabaseHaStrategy.PASSIVE_STANDBY,
                                        base_price * 2,
                                    )
                                )
                        for ha, ha_strategy, price in price_rows:
                            key = (
                                region.region_id,
                                database_id,
                                ha,
                                ha_strategy,
                            )
                            if key in seen:
                                continue
                            seen.add(key)
                            items.append(
                                {
                                    "vendor_id": vendor.vendor_id,
                                    "region_id": region.region_id,
                                    "database_id": database_id,
                                    "allocation": Allocation.ONDEMAND,
                                    "ha": ha,
                                    "ha_strategy": ha_strategy,
                                    "unit": PriceUnit.HOUR,
                                    "price": price,
                                    "price_upfront": 0,
                                    "price_tiered": [],
                                    "currency": currency,
                                }
                            )
        vendor.progress_tracker.advance_task()
    vendor.progress_tracker.hide_task()
    return items

inventory_database_storages

inventory_database_storages(vendor)

List all Azure Database for PostgreSQL Flexible Server storage options via the capabilities API.

Source code in sc_crawler/vendors/_azure.py

def inventory_database_storages(vendor):
    """List all Azure Database for PostgreSQL Flexible Server storage options via the capabilities API."""
    storages: dict[str, dict] = {}
    regions = _pg_database_regions(vendor)
    vendor.progress_tracker.start_task(
        name="Scanning region(s) for database_storage(s)", total=len(regions)
    )
    for region in regions:
        location = region.api_reference
        with sentry_capture_or_raise(vendor=vendor):
            for capability in _pg_capabilities(location):
                for edition in (
                    getattr(capability, "supported_server_editions", None) or []
                ):
                    for storage_edition in (
                        getattr(edition, "supported_storage_editions", None) or []
                    ):
                        if (
                            getattr(storage_edition, "reason", None)
                            == _PG_UNSUPPORTED_STORAGE_REASON
                        ):
                            continue
                        storage_id = storage_edition.name
                        if not storage_id:
                            continue

                        sizes_mb: list[int] = []
                        max_iops_values: list[int] = []
                        max_throughput = None
                        default_mb = getattr(
                            storage_edition, "default_storage_size_mb", None
                        )
                        if default_mb is not None:
                            sizes_mb.append(int(default_mb))
                        for storage_mb in storage_edition.supported_storage_mb or []:
                            size = getattr(storage_mb, "storage_size_mb", None)
                            if size is not None:
                                sizes_mb.append(int(size))
                            maximum_size = getattr(
                                storage_mb, "maximum_storage_size_mb", None
                            )
                            if maximum_size is not None:
                                sizes_mb.append(int(maximum_size))
                            iops = getattr(storage_mb, "supported_iops", None)
                            if iops is not None:
                                max_iops_values.append(int(iops))
                            maximum_iops = getattr(
                                storage_mb, "supported_maximum_iops", None
                            )
                            if maximum_iops is not None:
                                max_iops_values.append(int(maximum_iops))
                            for tier in (
                                getattr(storage_mb, "supported_iops_tiers", None) or []
                            ):
                                tier_iops = getattr(tier, "iops", None)
                                if tier_iops is not None:
                                    max_iops_values.append(int(tier_iops))
                            throughput = getattr(
                                storage_mb, "supported_maximum_throughput", None
                            )
                            if throughput is not None:
                                max_throughput = max(
                                    max_throughput or 0, int(throughput)
                                )

                        bounds = {
                            # Azure storage_size_mb values are MiB
                            "min_size": (
                                round(min(sizes_mb) / 1024 * _GIB_TO_GB)
                                if sizes_mb
                                else None
                            ),
                            "max_size": (
                                round(max(sizes_mb) / 1024 * _GIB_TO_GB)
                                if sizes_mb
                                else None
                            ),
                            "max_iops": (
                                max(max_iops_values) if max_iops_values else None
                            ),
                            "max_throughput": max_throughput,
                        }
                        merged = storages.setdefault(
                            storage_id,
                            {
                                "min_size": None,
                                "max_size": None,
                                "max_iops": None,
                                "max_throughput": None,
                            },
                        )
                        for key in (
                            "min_size",
                            "max_size",
                            "max_iops",
                            "max_throughput",
                        ):
                            left, right = merged.get(key), bounds.get(key)
                            if left is None:
                                merged[key] = right
                            elif right is not None:
                                merged[key] = (
                                    min(left, right)
                                    if key == "min_size"
                                    else max(left, right)
                                )
        vendor.progress_tracker.advance_task()
    vendor.progress_tracker.hide_task()

    items = []
    for storage_id, bounds in sorted(storages.items()):
        description_parts = [_PG_STORAGE_DESCRIPTIONS.get(storage_id, storage_id)]
        min_size = bounds.get("min_size")
        max_size = bounds.get("max_size")
        if min_size is not None and max_size is not None:
            description_parts.append(f"{min_size}-{max_size} GB")
        elif min_size is not None:
            description_parts.append(f"from {min_size} GB")
        elif max_size is not None:
            description_parts.append(f"up to {max_size} GB")
        if bounds.get("max_iops") is not None:
            description_parts.append(f"up to {bounds['max_iops']} IOPS")
        if bounds.get("max_throughput") is not None:
            description_parts.append(
                f"up to {bounds['max_throughput']} MB/s throughput"
            )
        items.append(
            {
                "vendor_id": vendor.vendor_id,
                "database_storage_id": storage_id,
                "name": storage_id,
                "description": ", ".join(description_parts),
                "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"),
            }
        )

    items.append(
        {
            "vendor_id": vendor.vendor_id,
            "database_storage_id": _PG_BACKUP_STORAGE_ID,
            "name": _PG_BACKUP_STORAGE_ID,
            "description": "Flexible Server backup storage (locally redundant)",
            # Backup SKU not in capabilities API; id matches retail Backup Storage LRS meter.
            # https://learn.microsoft.com/en-us/azure/postgresql/backup-restore/concepts-backup-restore
            "scope": DatabaseStorageScope.BACKUP,
            "redundancy": "LRS",
        }
    )
    return items

inventory_database_storage_prices

inventory_database_storage_prices(vendor)

Look up Azure Database for PostgreSQL Flexible Server storage prices via the Azure Retail Prices API.

For more information, see https://learn.microsoft.com/en-us/rest/api/cost-management/retail-prices/azure-retail-prices.

Source code in sc_crawler/vendors/_azure.py

def inventory_database_storage_prices(vendor):
    """Look up Azure Database for PostgreSQL Flexible Server storage prices via the Azure Retail Prices API.

    For more information, see <https://learn.microsoft.com/en-us/rest/api/cost-management/retail-prices/azure-retail-prices>.
    """
    items = []
    regions = _pg_database_regions(vendor)
    vendor.progress_tracker.start_task(
        name="Scanning region(s) for database_storage_price(s)",
        total=len(regions),
    )
    for region in regions:
        location = region.api_reference
        with sentry_capture_or_raise(vendor=vendor):
            supported_storage_ids = _pg_supported_storage_ids(location)
            for item in _pg_retail_prices(location):
                product = item.get("productName") or ""
                meter = (item.get("meterName") or "").lower()
                if product == _PG_FLEX_STORAGE_PRODUCT:
                    storage_id = _PG_STORAGE_RETAIL_TO_ID.get(meter)
                elif (
                    product == _PG_FLEX_BACKUP_PRODUCT
                    and meter == "backup storage lrs data stored"
                ):
                    storage_id = _PG_BACKUP_STORAGE_ID
                else:
                    storage_id = None
                if not storage_id:
                    continue
                if (
                    storage_id != _PG_BACKUP_STORAGE_ID
                    and storage_id not in supported_storage_ids
                ):
                    continue
                multiplier = STORAGE_PRICE_UNIT_MAPPING.get(item.get("unitOfMeasure"))
                if multiplier is None:
                    continue
                items.append(
                    {
                        "vendor_id": vendor.vendor_id,
                        "region_id": region.region_id,
                        "database_storage_id": storage_id,
                        "unit": PriceUnit.GB_MONTH,
                        "price": float(item["retailPrice"]) * multiplier,
                        "price_upfront": 0,
                        "price_tiered": [],
                        "currency": item.get("currencyCode", "USD"),
                    }
                )
        vendor.progress_tracker.advance_task()
    vendor.progress_tracker.hide_task()
    return items