gistfile1.txt

#!/usr/bin/env python3

import fileinput, sys, re
import requests, json
import argparse
import time


DEFAULT_BEARER_TOKEN = '<TOKEN>'


API_VERSION = '2021-10-01'
SUBSCRIPTION_ID = 'XXXXX'
START_DATE = '2025-12-14'
END_DATE = '2025-12-16'

URL = f'https://management.azure.com/subscriptions/{SUBSCRIPTION_ID}/providers/Microsoft.Consumption/usageDetails?$expand={EXPAND}&$top={BATCH_SIZE}&api-version={API_VERSION}&metric=actualcost&startDate={START_DATE}&endDate={END_DATE}'


def getRequest(token, url):
  headers = {'Authorization': f'Bearer {token}'}
  result = requests.get(url, headers=headers)
  return result

def saveChunk(fileNamePrefix, content, idx):
  fileName = f"{fileNamePrefix}.{str(idx).zfill(3)}.json"
  with open(fileName, mode='wb') as localfile:
    json_content = json.loads(content)
    output_content = json.dumps(json_content, indent=2)
    localfile.write(output_content.encode())

def getUrl(args):
  subscriptionId = args.subscriptionId
  expand = args.expand
  top = args.top
  apiVersion = args.apiVersion
  metric = args.metric# 'actualcost'
  start = args.startDate
  end = args.endDate
  return f'https://management.azure.com/subscriptions/{subscriptionId}/providers/Microsoft.Consumption/usageDetails?$expand={expand}&$top={top}&api-version={apiVersion}&metric={metric}&startDate={start}&endDate={end}'
  #return f'https://management.azure.com/subscriptions/{subscriptionId}/providers/Microsoft.Consumption/usageDetails?$top={top}&api-version={apiVersion}&metric={metric}&startDate={start}&endDate={end}'


def process(args):
  fileName = args.outputFilePrefix
  token = args.token # sys.argv[2] if len(sys.argv) > 2 else DEFAULT_BEARER_TOKEN

  url = getUrl(args)
  idx = 0
  while True:
    resp = getRequest(token, url)
    body = resp.json()
    if resp.status_code != 200:
      ##message = body['error']['message']
      print(body, file=sys.stderr)
      sys.exit(1)
    else:
      #print(resp.json())
      idx += 1
      saveChunk(fileName, resp.text, idx)
      if 'nextLink' not in body:
        print(f"Finished querying Azure Details after {idx} chunks")
        break
      else:
        url = body['nextLink']
        message = f"NEXT chunk: {url}" if args.sleep == 0 else f"Sleeping {args.sleep} seconds before next chunk: {url}"
        print(message)
        if args.sleep > 0:
          time.sleep(args.sleep)
      

if __name__ == "__main__":
  parser = argparse.ArgumentParser(description='Download modern Azure Usage Details')
  parser.add_argument('-s', '--subscriptionId', type=str, help='Azure SubscriptionId')
  parser.add_argument('-e', '--expand', default = 'meterDetails', type=str, help='expand argument')
  parser.add_argument('-a', '--apiVersion', default = '2023-11-01', type=str, help='API Version')
  parser.add_argument('-m', '--metric', default = 'actualcost', type=str, help='Metric Type: actualcost or amortizedcost')
  parser.add_argument('--startDate', type=str, required=True, help='start date')
  parser.add_argument('--endDate', type=str, required=True, help='end date')
  parser.add_argument('--top', default = 1000, type=int, help='top: batch size')
  parser.add_argument('-o', '--outputFilePrefix', required=False, default="usage_details_out", type=str, help='Prefix for output files')
  parser.add_argument('-t', '--token', type=str, help='Bearer Token')
  parser.add_argument('--sleep', type=int, help='Seconds to sleep between requests', default=0)

  args = parser.parse_args()
  process(args)