hetzner_pricing_generator.py (8206B)
1 #!/usr/bin/env python3 2 """ 3 Hetzner Cloud Pricing CSV Generator for OpenCost 4 5 This script fetches current pricing from the Hetzner Cloud API and generates 6 an OpenCost-compatible CSV file for cost tracking. 7 8 Requirements: 9 pip install requests 10 11 Usage: 12 # Set your Hetzner Cloud API token 13 export HCLOUD_TOKEN="your-api-token" 14 15 # Run the script 16 python3 hetzner_pricing_generator.py 17 18 # Output will be written to hetzner_pricing.csv 19 20 API Documentation: 21 https://docs.hetzner.cloud/reference/cloud#pricing 22 """ 23 24 import csv 25 import os 26 import sys 27 from datetime import datetime 28 from typing import Any 29 30 try: 31 import requests 32 except ImportError: 33 print("Error: 'requests' library is required. Install with: pip install requests") 34 sys.exit(1) 35 36 37 HETZNER_API_URL = "https://api.hetzner.cloud/v1" 38 39 # Hetzner regions 40 REGIONS = ["fsn1", "nbg1", "hel1", "ash", "hil", "sin"] 41 42 # EU regions (pricing in the CSV will use EU rates by default) 43 EU_REGIONS = ["fsn1", "nbg1", "hel1"] 44 45 46 def get_api_token() -> str: 47 """Get Hetzner Cloud API token from environment.""" 48 token = os.environ.get("HCLOUD_TOKEN") 49 if not token: 50 print("Error: HCLOUD_TOKEN environment variable not set") 51 print( 52 "Get your token from: https://console.hetzner.cloud/projects/*/security/tokens" 53 ) 54 sys.exit(1) 55 return token 56 57 58 def fetch_pricing(token: str) -> dict[str, Any]: 59 """Fetch pricing data from Hetzner Cloud API.""" 60 headers = {"Authorization": f"Bearer {token}"} 61 response = requests.get(f"{HETZNER_API_URL}/pricing", headers=headers) 62 63 if response.status_code != 200: 64 print(f"Error fetching pricing: {response.status_code}") 65 print(response.text) 66 sys.exit(1) 67 68 return response.json() 69 70 71 def fetch_server_types(token: str) -> dict[str, Any]: 72 """Fetch server types from Hetzner Cloud API.""" 73 headers = {"Authorization": f"Bearer {token}"} 74 response = requests.get(f"{HETZNER_API_URL}/server_types", headers=headers) 75 76 if response.status_code != 200: 77 print(f"Error fetching server types: {response.status_code}") 78 print(response.text) 79 sys.exit(1) 80 81 return response.json() 82 83 84 def fetch_load_balancer_types(token: str) -> dict[str, Any]: 85 """Fetch load balancer types from Hetzner Cloud API.""" 86 headers = {"Authorization": f"Bearer {token}"} 87 response = requests.get(f"{HETZNER_API_URL}/load_balancer_types", headers=headers) 88 89 if response.status_code != 200: 90 print(f"Error fetching load balancer types: {response.status_code}") 91 print(response.text) 92 sys.exit(1) 93 94 return response.json() 95 96 97 def parse_hourly_price(price_str: str) -> float: 98 """Parse price string to float.""" 99 try: 100 return float(price_str) 101 except (ValueError, TypeError): 102 return 0.0 103 104 105 def generate_csv(output_file: str = "hetzner_pricing.csv"): 106 """Generate OpenCost-compatible CSV from Hetzner API data.""" 107 token = get_api_token() 108 109 print("Fetching pricing data from Hetzner Cloud API...") 110 pricing_data = fetch_pricing(token) 111 server_types_data = fetch_server_types(token) 112 lb_types_data = fetch_load_balancer_types(token) 113 114 version = datetime.now().strftime("%Y.%m") 115 rows = [] 116 117 # Header 118 header = [ 119 "EndTimestamp", 120 "InstanceID", 121 "Region", 122 "AssetClass", 123 "InstanceIDField", 124 "InstanceType", 125 "MarketPriceHourly", 126 "Version", 127 ] 128 129 # Process server types 130 print("Processing server types...") 131 for server_type in server_types_data.get("server_types", []): 132 name = server_type["name"] 133 134 # Get pricing for each location 135 for price_info in server_type.get("prices", []): 136 location = price_info.get("location") 137 hourly_price = parse_hourly_price( 138 price_info.get("price_hourly", {}).get("net", "0") 139 ) 140 141 if hourly_price > 0: 142 rows.append( 143 [ 144 "", # EndTimestamp 145 name, # InstanceID 146 location, # Region 147 "node", # AssetClass 148 "metadata.labels.node.kubernetes.io/instance-type", # InstanceIDField 149 name, # InstanceType 150 f"{hourly_price:.6f}", # MarketPriceHourly 151 version, # Version 152 ] 153 ) 154 155 # Process load balancer types 156 print("Processing load balancer types...") 157 for lb_type in lb_types_data.get("load_balancer_types", []): 158 name = lb_type["name"] 159 160 for price_info in lb_type.get("prices", []): 161 location = price_info.get("location") 162 hourly_price = parse_hourly_price( 163 price_info.get("price_hourly", {}).get("net", "0") 164 ) 165 166 if hourly_price > 0: 167 rows.append( 168 [ 169 "", # EndTimestamp 170 name, # InstanceID 171 location, # Region 172 "node", # AssetClass (using node for LBs to track costs) 173 "metadata.labels.load-balancer.hetzner.cloud/type", # InstanceIDField 174 name, # InstanceType 175 f"{hourly_price:.6f}", # MarketPriceHourly 176 version, # Version 177 ] 178 ) 179 180 # Process volume pricing 181 print("Processing volume pricing...") 182 pricing = pricing_data.get("pricing", {}) 183 volume_pricing = pricing.get("volume", {}) 184 185 # Volume pricing is per GB per month, convert to hourly 186 price_per_gb_month_net = volume_pricing.get("price_per_gb_month", {}).get("net") 187 if price_per_gb_month_net is None: 188 # Fall back to a hardcoded default but warn the user that API data was unavailable 189 fallback_price_per_gb_month_net = "0.052" 190 print( 191 "Warning: Volume pricing (pricing.volume.price_per_gb_month.net) " 192 "not returned by Hetzner API; using fallback net price_per_gb_month=" 193 f"{fallback_price_per_gb_month_net} EUR/GB/month", 194 file=sys.stderr, 195 ) 196 price_per_gb_month_net = fallback_price_per_gb_month_net 197 volume_monthly = parse_hourly_price(price_per_gb_month_net) 198 volume_hourly = volume_monthly / 730 # Average hours per month 199 200 for region in EU_REGIONS: 201 rows.append( 202 [ 203 "", # EndTimestamp 204 "hcloud-volumes", # InstanceID 205 region, # Region 206 "pv", # AssetClass 207 "spec.storageClassName", # InstanceIDField 208 "hcloud-volumes", # InstanceType 209 f"{volume_hourly:.8f}", # MarketPriceHourly 210 version, # Version 211 ] 212 ) 213 214 # Write CSV 215 print(f"Writing {len(rows)} pricing entries to {output_file}...") 216 with open(output_file, "w", newline="") as f: 217 # Write header comment 218 f.write("# Hetzner Cloud Pricing for OpenCost CSV Provider\n") 219 f.write(f"# Generated: {datetime.now().isoformat()}\n") 220 f.write("# Source: Hetzner Cloud API (https://api.hetzner.cloud/v1/pricing)\n") 221 f.write("# Prices in EUR per hour (net, excluding VAT)\n") 222 f.write("#\n") 223 224 writer = csv.writer(f) 225 writer.writerow(header) 226 writer.writerows(rows) 227 228 print(f"Done! Generated {output_file} with {len(rows)} pricing entries.") 229 print("\nTo use with OpenCost:") 230 print(" 1. Copy the CSV to your OpenCost deployment") 231 print(" 2. Set USE_CSV_PROVIDER=true") 232 print(" 3. Set CSV_PATH to the file location") 233 234 235 def main(): 236 """Main entry point.""" 237 import argparse 238 239 parser = argparse.ArgumentParser( 240 description="Generate OpenCost-compatible pricing CSV from Hetzner Cloud API" 241 ) 242 parser.add_argument( 243 "-o", 244 "--output", 245 default="hetzner_pricing.csv", 246 help="Output CSV file path (default: hetzner_pricing.csv)", 247 ) 248 249 args = parser.parse_args() 250 generate_csv(args.output) 251 252 253 if __name__ == "__main__": 254 main()