_azure

sc_crawler.vendors._azure

Inventory of Microsoft Azure via its APIs: servers, storages, traffic, IPv4 addresses and managed PostgreSQL (Flexible Server), with their prices.

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 Azure Database for PostgreSQL Flexible Server types in all regions.
inventory_database_prices	List Azure Database for PostgreSQL Flexible Server on-demand prices from the 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.

Lifecycle: _azure_sku_lifecycle_status checks _AZURE_SKU_LIFECYCLE_RULES — RETIRED if the retirement date has passed, PLANNED_FOR_RETIREMENT if matched but not yet retired, otherwise ACTIVE. Same function is used for database SKUs.

Source code in sc_crawler/vendors/_azure.py

def inventory_servers(vendor):
    """List all available instance types in all regions.

    Lifecycle: `_azure_sku_lifecycle_status` checks `_AZURE_SKU_LIFECYCLE_RULES`
    — RETIRED if the retirement date has passed, PLANNED_FOR_RETIREMENT if
    matched but not yet retired, otherwise ACTIVE. Same function is used for
    database SKUs.
    """
    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 = _get_storage_retail_prices()
    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 = _get_inbound_traffic_retail_prices()
        outbound_prices = _get_outbound_traffic_retail_prices()
    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 = _get_ipv4_retail_prices()
    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 Azure Database for PostgreSQL Flexible Server types in all regions.

Retail compute meters decide which instances exist. Capabilities is quota-filtered, so a size can be missing there. When database_id is present, capabilities supplies vCPUs, memory, HA, engine versions, storage bounds, and autotuning. Otherwise those come from another SKU of the same edition, and vCPUs/memory come from the matching VM. A per-vCore series meter has no size list; matching VM SKUs name the sizes.

Source code in sc_crawler/vendors/_azure.py

def inventory_databases(vendor):
    """List Azure Database for PostgreSQL Flexible Server types in all regions.

    Retail compute meters decide which instances exist. Capabilities is
    quota-filtered, so a size can be missing there. When ``database_id`` is
    present, capabilities supplies vCPUs, memory, HA, engine versions,
    storage bounds, and autotuning. Otherwise those come from another SKU of
    the same edition, and vCPUs/memory come from the matching VM.
    A per-vCore series meter has no size list; matching VM SKUs name the sizes.
    """
    # database_id -> fields that actually come from capabilities.
    caps_by_id: dict[str, dict] = {}
    edition_caps: dict[str, dict] = {}
    retail_concrete: dict[str, str] = {}
    retail_series: dict[str, str] = {}
    servers_by_id = {}
    servers_by_series: dict[str, list] = defaultdict(list)
    for server in vendor.servers:
        name = server.api_reference
        if not name or "-" in name:
            continue
        servers_by_id[name] = server
        match = _PG_DATABASE_NAME_RE.match(name)
        # i/m/l/t are stripped from the retail series name, so those VM shapes are not this series.
        if match and not _PG_SERIES_SKIP_FEATURES & set(match["features"].lower()):
            servers_by_series[_parse_database_name(name)[0]].append(server)

    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):
                features = {
                    getattr(feature, "name", None): getattr(feature, "status", None)
                    == "Enabled"
                    for feature in getattr(capability, "supported_features", None) or []
                }
                shared = {
                    "engine_versions": _pg_engine_versions(capability),
                    "storage_extra_autosize": {"Enabled": True, "Disabled": False}.get(
                        getattr(capability, "storage_auto_growth_supported", None),
                        features.get("StorageAutoGrowth"),
                    ),
                    # IndexTuning → advice; AdaptiveAutoVacuumAutoApply → apply (vacuum GUCs).
                    # ConfigTuning may be Writes tuning, but it's not documented.
                    # https://learn.microsoft.com/en-us/azure/postgresql/monitor/concepts-autonomous-tuning
                    # https://learn.microsoft.com/en-us/azure/postgresql/monitor/concepts-adaptive-autovacuum
                    # https://learn.microsoft.com/en-us/azure/postgresql/monitor/concepts-intelligent-tuning
                    "autotuning_advice": features.get("IndexTuning", False),
                    "autotuning_apply": features.get(
                        "AdaptiveAutoVacuumAutoApply", False
                    ),
                }
                for edition in (
                    getattr(capability, "supported_server_editions", None) or []
                ):
                    # Azure storage_size_mb values are MiB.
                    # Premium SSD v2 is General Purpose and Memory Optimized only, Burstable is capped by Premium SSD (~32 TiB).
                    # https://learn.microsoft.com/en-us/azure/postgresql/compute-storage/concepts-storage-premium-ssd-v2
                    sizes_gb = [
                        round(int(size_mb) / 1024 * _GIB_TO_GB)
                        for storage_edition in (
                            getattr(edition, "supported_storage_editions", None) or []
                        )
                        if getattr(storage_edition, "reason", None)
                        != _PG_UNSUPPORTED_STORAGE_REASON
                        and not (
                            edition.name == "Burstable"
                            and storage_edition.name == "ManagedDiskV2"
                        )
                        for storage_mb in (
                            getattr(storage_edition, "supported_storage_mb", None) or []
                        )
                        for size_mb in (
                            getattr(storage_mb, "storage_size_mb", None),
                            getattr(storage_mb, "maximum_storage_size_mb", None),
                        )
                        if size_mb is not None
                    ]
                    shared["storage_extra_min"] = min(sizes_gb, default=None)
                    shared["storage_extra_max"] = max(sizes_gb, default=None)
                    for sku in getattr(edition, "supported_server_skus", None) or []:
                        # https://learn.microsoft.com/en-us/azure/postgresql/high-availability/concepts-high-availability
                        # Burstable does not support HA (the API may still list modes).
                        ha_modes = {
                            getattr(mode, "value", str(mode))
                            for mode in getattr(sku, "supported_ha_mode", None) or []
                        }
                        ha = [DatabaseHaLevel.NONE]
                        if edition.name != "Burstable":
                            if "ZoneRedundant" in ha_modes:
                                ha.append(DatabaseHaLevel.MULTI_ZONE)
                            if "SameZone" in ha_modes:
                                ha.append(DatabaseHaLevel.SINGLE_ZONE)
                        ha_strategy = [DatabaseHaStrategy.NONE]
                        if len(ha) > 1:
                            ha_strategy.append(DatabaseHaStrategy.PASSIVE_STANDBY)
                        previous = caps_by_id.get(sku.name)
                        if previous and DatabaseHaLevel.MULTI_ZONE in previous["ha"]:
                            ha = previous["ha"]
                        else:
                            ha = DatabaseHaLevel.ordered(ha)
                        facts = {
                            **shared,
                            "edition": edition.name,
                            "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
                            ),
                            "ha": ha,
                            "ha_strategy": DatabaseHaStrategy.ordered(ha_strategy),
                            # https://learn.microsoft.com/en-us/azure/reliability/reliability-database-postgresql
                            "sla": (
                                99.99
                                if DatabaseHaLevel.MULTI_ZONE in ha
                                else 99.95
                                if DatabaseHaLevel.SINGLE_ZONE in ha
                                else 99.9
                            ),
                        }
                        caps_by_id[sku.name] = facts
                        edition_caps[edition.name] = facts
        with sentry_capture_or_raise(vendor=vendor):
            for item in _pg_retail_prices(location):
                product = item.get("productName", "")
                product_l = product.lower()
                if (
                    "compute" not in product_l
                    or "cosmos" in product_l
                    or "single server" in product_l
                    or not (
                        "flexible server" in product_l or "flex server" in product_l
                    )
                ):
                    continue
                arm = item.get("armSkuName", "")
                series_match = _PG_PRODUCT_SERIES_RE.search(product)
                series_name = series_match["series"] if series_match else ""
                edition_name = next(
                    (
                        edition
                        for edition, tokens in _PG_EDITION_PRODUCT_TOKENS.items()
                        if any(token in product_l for token in tokens)
                    ),
                    "MemoryOptimized"
                    if series_name.startswith(("E", "M"))
                    else "GeneralPurpose",
                )
                if arm.startswith("Standard_") or arm.upper() in _PG_FLAT_COMPUTE_SKUS:
                    match = _PG_DATABASE_NAME_RE.match(arm)
                    if match is None:
                        continue
                    database_id = (
                        f"Standard_{match['family'].upper()}"
                        f"{match['vcpus']}{match['features'].lower()}"
                    )
                    if match["version"]:
                        database_id += f"_v{match['version']}"
                    retail_concrete.setdefault(database_id, edition_name)
                elif series_name:
                    retail_series.setdefault(series_name, edition_name)
        vendor.progress_tracker.advance_task()
    vendor.progress_tracker.hide_task()

    # Capabilities SKUs, plus retail SKUs quota hid. Series meters name no size,
    # so the VM catalog fills those database_ids.
    wanted = {
        database_id: facts["edition"] for database_id, facts in caps_by_id.items()
    }
    for database_id, edition_name in retail_concrete.items():
        wanted.setdefault(database_id, edition_name)
    present_series = {_parse_database_name(database_id)[0] for database_id in wanted}
    for series_name, edition_name in retail_series.items():
        if series_name not in present_series:
            for server in servers_by_series.get(series_name, []):
                wanted.setdefault(server.api_reference, edition_name)

    rows = []
    for database_id, edition_name in wanted.items():
        own = caps_by_id.get(database_id)
        facts = (
            own
            or edition_caps.get(edition_name)
            or next(iter(edition_caps.values()), None)
        )
        if facts is None:
            continue
        server = servers_by_id.get(database_id)
        series, _family, vcpus, _version = _parse_database_name(database_id)
        ha, ha_strategy, sla = facts["ha"], facts["ha_strategy"], facts["sla"]
        if own:
            edition_name = own["edition"]
            vcpus, memory_amount = own["vcpus"], own["memory_amount"]
        else:
            if server and server.vcpus:
                vcpus = server.vcpus
            memory_amount = (
                int(server.memory_amount) if server and server.memory_amount else None
            )
            if edition_name == "Burstable" and edition_name not in edition_caps:
                ha, ha_strategy, sla = (
                    [DatabaseHaLevel.NONE],
                    [DatabaseHaStrategy.NONE],
                    99.9,
                )
        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" ({', '.join(spec_parts)})"
        name = database_id.removeprefix("Standard_")
        sku_prefix = _PG_SKU_NAME_PREFIX.get(edition_name)
        rows.append(
            {
                "vendor_id": vendor.vendor_id,
                "database_id": database_id,
                "name": name,
                "api_reference": database_id,
                # https://www.pulumi.com/registry/packages/azure/api-docs/postgresql/flexibleserver/
                "api_reference_object": {
                    "sku_name": (
                        f"{sku_prefix}_{database_id}" if sku_prefix else database_id
                    )
                },
                "display_name": name,
                "description": description,
                "server_id": server.server_id if server else None,
                "engine": DatabaseEngine.POSTGRESQL,
                "wire_protocol": DatabaseWireProtocol.POSTGRESQL,
                "engine_versions": facts["engine_versions"],
                "family": series,
                "vcpus": vcpus,
                "memory_amount": memory_amount,
                "storage_size": None,
                "storage_extra_autosize": facts["storage_extra_autosize"],
                # https://learn.microsoft.com/en-us/azure/postgresql/backup-restore/concepts-backup-restore
                "scheduled_backups": True,
                "autotuning_advice": facts["autotuning_advice"],
                "autotuning_apply": facts["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
                "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": facts["storage_extra_min"],
                "storage_extra_max": facts["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
                "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
                "max_read_replicas": 0 if edition_name == "Burstable" else 5,
                # https://learn.microsoft.com/en-us/azure/postgresql/connectivity/concepts-pgbouncer
                "connection_pool": True,
                # https://learn.microsoft.com/en-us/azure/postgresql/monitor/concepts-monitoring
                "system_monitoring": True,
                # https://learn.microsoft.com/en-us/azure/postgresql/monitor/concepts-query-performance-insight
                "database_monitoring": True,
                "continuous_backups": 35,
                "sla": sla,
                "status": _azure_sku_lifecycle_status(database_id),
            }
        )
    return rows

inventory_database_prices

inventory_database_prices(vendor)

List Azure Database for PostgreSQL Flexible Server on-demand prices from the Retail Pricing API.

SKU catalog comes from vendor.databases (built by inventory_databases / capabilities). Per-region capabilities are not used for pricing: emit a row whenever Retail has a meter for a known instance in that region.

OfferRestricted=Enabled means this subscription cannot list or provision Flexible Server SKUs there (empty supported_server_editions), so capabilities give no confirmed region availability. Retail list prices may still exist; we keep those rows. https://learn.microsoft.com/en-us/azure/postgresql/troubleshoot/how-to-resolve-capacity-errors https://learn.microsoft.com/en-us/rest/api/postgresql/capabilities-by-location/list

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 Azure Database for PostgreSQL Flexible Server on-demand prices from the Retail Pricing API.

    SKU catalog comes from ``vendor.databases`` (built by ``inventory_databases`` /
    capabilities). Per-region capabilities are not used for pricing: emit a row
    whenever Retail has a meter for a known instance in that region.

    OfferRestricted=Enabled means this subscription cannot list or provision
    Flexible Server SKUs there (empty supported_server_editions), so capabilities
    give no confirmed region availability. Retail list prices may still exist;
    we keep those rows.
    https://learn.microsoft.com/en-us/azure/postgresql/troubleshoot/how-to-resolve-capacity-errors
    https://learn.microsoft.com/en-us/rest/api/postgresql/capabilities-by-location/list

    More information: <https://learn.microsoft.com/en-us/rest/api/cost-management/retail-prices/azure-retail-prices>.
    """
    items = []
    seen = 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:
        with sentry_capture_or_raise(vendor=vendor):
            prices_by_arm: dict[str, list[dict]] = defaultdict(list)
            for item in _pg_retail_prices(region.api_reference):
                arm = item.get("armSkuName", "")
                if arm and (
                    "compute" in item.get("productName", "").lower()
                    or arm.startswith("Standard_")
                    or arm.upper() in _PG_FLAT_COMPUTE_SKUS
                    or item.get("meterName", "").upper() in _PG_FLAT_COMPUTE_SKUS
                ):
                    prices_by_arm[arm].append(item)

            for database in vendor.databases:
                if not database.status.is_orderable:
                    continue
                price_item = _pg_lookup_retail_price(
                    vendor=vendor,
                    database_id=database.database_id,
                    edition_name=database.family,
                    prices_by_arm=prices_by_arm,
                )
                if price_item is None:
                    continue
                base_price = _pg_hourly_compute_price(
                    price_item, database.vcpus, database_id=database.database_id
                )
                # 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_levels = {DatabaseHaLevel(level) for level in database.ha or []}
                price_rows = [
                    (DatabaseHaLevel.NONE, DatabaseHaStrategy.NONE, base_price)
                ] + [
                    (level, DatabaseHaStrategy.PASSIVE_STANDBY, base_price * 2)
                    for level in (
                        DatabaseHaLevel.MULTI_ZONE,
                        DatabaseHaLevel.SINGLE_ZONE,
                    )
                    if level in ha_levels
                ]
                for ha, ha_strategy, price in price_rows:
                    key = (region.region_id, database.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.database_id,
                            "allocation": Allocation.ONDEMAND,
                            "ha": ha,
                            "ha_strategy": ha_strategy,
                            "unit": PriceUnit.HOUR,
                            "price": price,
                            "price_upfront": 0,
                            "price_tiered": [],
                            "currency": price_item.get("currencyCode", "USD"),
                        }
                    )
        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 "").lower()
                meter = (item.get("meterName") or "").lower()
                storage_id = (
                    _PG_STORAGE_RETAIL_TO_ID.get(meter)
                    if product in _PG_STORAGE_RETAIL_PRODUCTS
                    else 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