概要

Azure上の指定したサブスクリプションの請求データを取得して、Azure Blob Storage に Export するPythonプログラムです。本プログラムは CostManagement REST API を利用して作成しています。

請求データをExportするには以下のカテゴリがあり、必要に応じてコールします。

list : Exportクエリ一覧の取得（スコープ内容の取得）
create : Exportクエリの作成（請求データのクエリ定義）
execute : Exportクエリの実行（請求データの取得とBlobへの保存）
delete : Exportクエリの削除
history : Exportクエリの実行履歴の取得
get : Exportクエリ名でのスコープ内容の取得

実行環境

macOS Monterey 12.3.1
python 3.8.12
azure cli 2.41.0

事前準備

請求データを保存する Azure Blob Storage のストレージアカウント・リソースIDを取得します（参照）。

$ az storage account show --name [StorageAccountName] --resource-group [ResourceGroupName] --query id
"/subscriptions/[SubscriptionId]/resourceGroups/[ResourceGroupName]/providers/Microsoft.Storage/storageAccounts/[StorageAccountName]"

Pythonプログラムを実行する権限のあるアカウントのアクセストークンを取得します。一時的な環境変数として定義しておきます。有効期限は約１時間ですので、失効した場合は再実行ください

$ export EXP_TOKEN=$(az account get-access-token --query 'accessToken' --output tsv)

実行プログラム

デフォルトのExportクエリ名は「TestExport」、請求データ取得期間は「2022-09-01 〜 2022-09-15」です。

EXPORT_NwHubBySubscriptionId.py

import json
import os
import sys
import requests
import argparse
from datetime import *
import time
import logging

# export EXP_TOKEN=$(az account get-access-token --query 'accessToken' --output tsv)
ACCESS_TOKEN = os.environ['EXP_TOKEN']

# Exportクエリ情報
API_URL = 'https://management.azure.com'
API_VERSION = '?api-version=2022-10-01'
PV_EXPORT = '/providers/Microsoft.CostManagement/exports'

# デリバリ情報（保存先BLOBストレージ情報）
STORAGE_ACCOUNT_RESOURCE_ID = '/subscriptions/[SubscriptionId]/resourceGroups/[ResourceGroupName]/providers/Microsoft.Storage/storageAccounts/[StorageAccountName]'
CONTAINER_NAME = "partnercenter"
ROOT_FOLDER_PATH = "billing-data"

# Subscription ID
SUBSCRIPTION_ID = 'eeeeeeee-1717-4343-9797-ffffffffffff'

# SCOPEの定義
SCOPE = f'/subscriptions/{SUBSCRIPTION_ID}'


# クエリの定義（スケジュール定義なし）
def define_export_usage_payload(fm_tp, to_tp) :
    payload = {
        "properties": {
            "format": "Csv",
            "deliveryInfo": {
                "destination": {
                    "resourceId": STORAGE_ACCOUNT_RESOURCE_ID,
                    "container": CONTAINER_NAME,
                    "rootFolderPath": ROOT_FOLDER_PATH
                }
            },
            "definition": {
                "type": "ActualCost",
                "timeframe": "Custom",
                "timePeriod": {"from": fm_tp, "to": to_tp},
                "dataSet": {
                    "granularity": "Daily",
                    "configuration": {
                        "columns": [
                            "Date",
                            "ServiceFamily",
                            "ResourceId",
                            "ProductName",
                            "Quantity"
                        ]
                    }
                }
            }
        }
    }
    return payload


# Exportクエリ一覧の取得
def requests_list(headers):
    # URLの定義
    URL = f'{API_URL}{SCOPE}{PV_EXPORT}{API_VERSION}'
    print("\n EXPORT LIST : \n", URL)

    # REST-API 
    response = requests.get(
        URL,
        headers=headers
    )
    # requrest処理のクローズ
    response.close

    # リクエスト結果の確認
    if response_status_check(response):
        print(json.dumps(response.json()['value'], indent=2))


# Exportクエリの作成
def requests_create(headers, name, fm_tp, to_tp):
    # URLの定義
    URL = f'{API_URL}{SCOPE}{PV_EXPORT}{"/"}{name}{API_VERSION}'
    print("\n EXPORT CREATE : \n", URL)

    # 取得データ日付の確認
    if check_time_period(fm_tp, to_tp) == False :
        print("NG", "\n")
        return

    # REST-API データ部の作成（Exportクエリの定義）
    payload = define_export_usage_payload(fm_tp, to_tp)

    # REST-API 
    response = requests.put(
        URL,
        headers=headers,
        data=json.dumps(payload)
    )
    # requrest処理のクローズ
    response.close
    response_status_check(response)


# Exportクエリの実行
def requests_execute(headers, name):
    # URLの定義
    URL = f'{API_URL}{SCOPE}{PV_EXPORT}{"/"}{name}{"/run"}{API_VERSION}'
    print("\n EXPORT EXECUTE : \n", URL)
    response = requests.post(
        URL,
        headers=headers
    )
    # requrest処理のクローズ
    response.close
    response_status_check(response)


# Exportクエリの削除
def requests_delete(headers, name):
    # URLの定義
    URL = f'{API_URL}{SCOPE}{PV_EXPORT}{"/"}{name}{API_VERSION}'
    print("\n EXPORT DELETE : \n", URL)
    response = requests.delete(
        URL,
        headers=headers
    )
    # requrest処理のクローズ
    response.close
    response_status_check(response)


# Exportクエリの実行履歴の取得
def requests_history(headers, name):
    # URLの定義
    URL = f'{API_URL}{SCOPE}{PV_EXPORT}{"/"}{name}{"/runHistory"}{API_VERSION}'
    print("\n EXPORT HISTORY : \n", URL)
    response = requests.get(
        URL,
        headers=headers
    )
    # requrest処理のクローズ
    response.close

    # リクエスト結果の確認
    if response_status_check(response):
        print(json.dumps(response.json()['value'], indent=2))


# Exportクエリ名でのスコープ内容の取得
def requests_get(headers, name):
    # URLの定義
    URL = f'{API_URL}{SCOPE}{PV_EXPORT}{"/"}{name}{API_VERSION}'
    print("\n EXPORT GET : \n", URL)
    response = requests.get(
        URL,
        headers=headers
    )
    # requrest処理のクローズ
    response.close

    # リクエスト結果の確認
    if response_status_check(response):
        print(json.dumps(response.json(), indent=2))


# REST-APIのレスポンスチェック
def response_status_check(response):
    try :
        print(response)
        response.raise_for_status()
        return True
    except requests.exceptions.RequestException as e :
        logging.exception("export_cost_by_subscriptionid request failed. message=(%s)\n", e.response.text)
        sys.exit()


# SubscriptionIdでの請求データのExport処理
def export_cost_by_subscriptionid(fm_tp, to_tp, name, job) :

    # 取得するためのヘッダ情報
    headers = {
        'accept': '*/*',
        'Content-Type': 'application/json',
        'Authorization': 'Bearer %s' % ACCESS_TOKEN
    }

    if job == 'list':
        requests_list(headers)
    elif job == 'create':
        requests_create(headers, name, fm_tp, to_tp)
    elif job == 'execute':
        requests_execute(headers, name)
    elif job == 'delete':
        requests_delete(headers, name)
    elif job == 'history':
        requests_history(headers, name)
    elif job == 'get':
        requests_get(headers, name)
    else:
        print("\n\t define job : list / create / execute / delete / history / get")


# データ取得のための From-To 日付確認
def check_time_period(fm_tp, to_tp):
    try :
        fm_datetime = datetime.strptime(fm_tp, '%Y-%m-%d')
        to_datetime = datetime.strptime(to_tp, '%Y-%m-%d')

        if to_datetime < fm_datetime :
            print("Error ---> 開始日 ≦ 終了日 ")
            return False
        else :
            return True
    except ValueError as e :
        print("ValueError = ", e)
        return False


if __name__ == '__main__':
    parser = argparse.ArgumentParser(description='あるサブスクリプションの請求情報の取得')
    parser.add_argument('-f', '--fm', type=str, default='2022-09-01', help='データを取得する開始日(yyyy-mm-dd)')
    parser.add_argument('-t', '--to', type=str, default='2022-09-15', help='データを取得する終了日(yyyy-mm-dd)')
    # parser.add_argument('-f', '--fm', type=str, required=True, help='データを取得する開始日(yyyy-mm-dd)')
    # parser.add_argument('-t', '--to', type=str, required=True, help='データを取得する終了日(yyyy-mm-dd)')
    parser.add_argument('-n', '--name', type=str, default='TestExport', help='Exportクエリ名')
    parser.add_argument('-j', '--job', type=str, default='list', help='list / create / exeute / delete / history / get')
    args = parser.parse_args()

    start = time.time()
    export_cost_by_subscriptionid(args.fm, args.to, args.name, args.job)
    generate_time = time.time() - start
    print("\n処理時間:{0}".format(generate_time) + " [sec]\n")

プログラムのHELP表示

$ python EXPORT_NwHubBySubscriptionId.py -h     
usage: EXPORT_NwHubBySubscriptionId.py [-h] [-f FM] [-t TO] [-n NAME] [-j JOB]

あるサブスクリプションの請求情報の取得

optional arguments:
  -h, --help            show this help message and exit
  -f FM, --fm FM        データを取得する開始日(yyyy-mm-dd)
  -t TO, --to TO        データを取得する終了日(yyyy-mm-dd)
  -n NAME, --name NAME  Exportクエリ名
  -j JOB, --job JOB     list / create / exeute / delete / history / get

プログラムの実行

初期状態の確認、Exportクエリ一覧の取得

$ python EXPORT_NwHubBySubscriptionId.py                                            

 EXPORT LIST : 
 https://management.azure.com/subscriptions/eeeeeeee-1717-4343-9797-ffffffffffff/providers/Microsoft.CostManagement/exports?api-version=2022-10-01
<Response [200]>
[]

Exportクエリの作成

$ python EXPORT_NwHubBySubscriptionId.py -j create 

 EXPORT CREATE : 
 https://management.azure.com/subscriptions/eeeeeeee-1717-4343-9797-ffffffffffff/providers/Microsoft.CostManagement/exports/TestExport?api-version=2022-10-01
<Response [201]>

処理時間:9.942291975021362 [sec]

作成したExportクエリ名でのスコープ内容の取得

$ python EXPORT_NwHubBySubscriptionId.py -j get

 EXPORT GET : 
 https://management.azure.com/subscriptions/eeeeeeee-1717-4343-9797-ffffffffffff/providers/Microsoft.CostManagement/exports/TestExport?api-version=2022-10-01
<Response [200]>
{
  "id": "subscriptions/eeeeeeee-1717-4343-9797-ffffffffffff/providers/Microsoft.CostManagement/exports/TestExport",
  "name": "TestExport",
  "type": "Microsoft.CostManagement/exports",
  "eTag": "\"1d8e8d644c3d6f7\"",
  "properties": {
    "schedule": {
      "status": "Inactive"
    },
    "format": "Csv",
    "deliveryInfo": {
      "destination": {
        "resourceId": "/subscriptions/eeeeeeee-1717-4343-9797-ffffffffffff/resourceGroups/[ResourceGroupName]/providers/Microsoft.Storage/storageAccounts/[StorageAccountName]",
        "container": "partnercenter",
        "rootFolderPath": "billing-data"
      }
    },
    "definition": {
      "type": "ActualCost",
      "timeframe": "Custom",
      "timePeriod": {
        "from": "2022-09-01T00:00:00Z",
        "to": "2022-09-15T00:00:00Z"
      },
      "dataSet": {
        "configuration": {
          "columns": [
            "Date",
            "ServiceFamily",
            "ResourceId",
            "ProductName",
            "Quantity"
          ]
        },
        "granularity": "Daily"
      }
    }
  }
}

処理時間:3.424717903137207 [sec]

Exportクエリの実行

$ python EXPORT_NwHubBySubscriptionId.py -j execute

 EXPORT EXECUTE : 
 https://management.azure.com/subscriptions/eeeeeeee-1717-4343-9797-ffffffffffff/providers/Microsoft.CostManagement/exports/TestExport/run?api-version=2022-10-01
<Response [200]>

処理時間:5.772066116333008 [sec]

Exportクエリの実行履歴の取得

$ python EXPORT_NwHubBySubscriptionId.py -j history

 EXPORT HISTORY : 
 https://management.azure.com/subscriptions/eeeeeeee-1717-4343-9797-ffffffffffff/providers/Microsoft.CostManagement/exports/TestExport/runHistory?api-version=2022-10-01
<Response [200]>
[
  {
    "id": "subscriptions/eeeeeeee-1717-4343-9797-ffffffffffff/providers/Microsoft.CostManagement/exports/TestExport/Run/737d2db4-8e96-4dcb-b17e-3e33d6429121",
    "name": null,
    "type": null,
    "eTag": null,
    "properties": {
      "runSettings": {
        "format": "Csv",
        "deliveryInfo": {
          "destination": {
            "resourceId": "/subscriptions/eeeeeeee-1717-4343-9797-ffffffffffff/resourceGroups/[ResourceGroupName]/providers/Microsoft.Storage/storageAccounts/[StorageAccountName]",
            "container": "partnercenter",
            "rootFolderPath": "billing-data"
          }
        },
        "definition": {
          "type": "ActualCost",
          "timeframe": "Custom",
          "timePeriod": {
            "from": "2022-09-01T00:00:00Z",
            "to": "2022-09-15T00:00:00Z"
          },
          "dataSet": {
            "configuration": {
              "columns": [
                "Date",
                "ServiceFamily",
                "ResourceId",
                "ProductName",
                "Quantity"
              ]
            },
            "granularity": "Daily"
          }
        }
      },
      "executionType": "OnDemand",
      "status": "Completed",
      "submittedBy": "itsuruzono@networld.co.jp",
      "submittedTime": "2022-10-26T01:25:02.1987319Z",
      "processingStartTime": "2022-10-26T01:25:03.5328513Z",
      "processingEndTime": "2022-10-26T01:25:24.0105159Z",
      "fileName": "billing-data/TestExport/20220901-20220915/TestExport_737d2db4-8e96-4dcb-b17e-3e33d6429121.csv"
    }
  }
]

処理時間:1.5970940589904785 [sec]

Exportクエリの削除

$ python EXPORT_NwHubBySubscriptionId.py -j delete

 EXPORT DELETE : 
 https://management.azure.com/subscriptions/eeeeeeee-1717-4343-9797-ffffffffffff/providers/Microsoft.CostManagement/exports/TestExport?api-version=2022-10-01
<Response [200]>

処理時間:2.709244966506958 [sec]

実行後の確認

Blob を Azure cli から確認します。

$ az storage blob list --account-name [StorageAccountName] --container-name partnercenter -o table
Name                                                                                           Blob Type    Blob Tier    Length    Content Type              Last Modified              Snapshot
---------------------------------------------------------------------------------------------  -----------  -----------  --------  ------------------------  -------------------------  ----------
billing-data                                                                                   BlockBlob    Hot                                              2022-10-26T01:25:20+00:00
billing-data/TestExport                                                                        BlockBlob    Hot                                              2022-10-26T01:25:20+00:00
billing-data/TestExport/20220901-20220915                                                      BlockBlob    Hot                                              2022-10-26T01:25:20+00:00
billing-data/TestExport/20220901-20220915/TestExport_737d2db4-8e96-4dcb-b17e-3e33d6429121.csv  BlockBlob    Hot          820610    application/octet-stream  2022-10-26T01:25:23+00:00

このPythonプログラムを利用しても確認できます。

$ python blob_list.py                             
Size:0   	 Name:billing-data 
Size:0   	 Name:billing-data/TestExport 
Size:0   	 Name:billing-data/TestExport/20220901-20220915 
Size:820610   	 Name:billing-data/TestExport/20220901-20220915/TestExport_737d2db4-8e96-4dcb-b17e-3e33d6429121.csv

Blobデータのダウンロード

Blob Storage に保存された請求データをローカルにダウンロードするには、このPythonプログラムを参照利用ください。

おまけ

プログラム実行時に下記エラーが表示されたら、、、、

ERROR:root:export_cost_by_subscriptionid request failed. message=({"error":{"code":"400","message":"RP Not Registered. Register destination storage account subscription with Microsoft.CostManagementExports. Please refer https://docs.microsoft.com/en-us/rest/api/resources/providers/register"}})

該当サブスクリプションに「Microsoft.CostManagementExports」をプロバイダー登録ください。

$ az account set --subscription [SubscriptionName]
$ az provider register --namespace Microsoft.CostManagementExports --wait

付録

クエリ定義の configuration columns で指定できる項目一覧

'InvoiceId','PreviousInvoiceId',
'BillingAccountId','BillingAccountName','BillingProfileId','BillingProfileName',
'InvoiceSectionId','InvoiceSectionName',
'ResellerName','ResellerMPNId','CostCenter',
'BillingPeriodEndDate','BillingPeriodStartDate',
'ServicePeriodEndDate','ServicePeriodStartDate','Date','ServiceFamily',
'ProductOrderId','ProductOrderName','ConsumedService',
'MeterId','MeterName','MeterCategory','MeterSubcategory','MeterRegion',
'ProductId','ProductName',
'SubscriptionId','SubscriptionName',
'PublisherType','PublisherId','PublisherName',
'ResourceGroup','ResourceId','ResourceLocation','ResourceLocationNormalized',
'EffectivePrice','Quantity','UnitOfMeasure','ChargeType',
'BillingCurrencyCode','PricingCurrencyCode','CostInBillingCurrency','CostInPricingCurrency',
'CostInUSD','PaygCostInBillingCurrency','PaygCostInUSD',
'ExchangeRate','ExchangeRateDate','IsAzureCreditEligible',
'ServiceInfo1','ServiceInfo2','AdditionalInfo',
'Tags','PayGPrice','Frequency','Term',
'ReservationId','ReservationName','PricingModel','UnitPrice',
'CostAllocationRuleName','BenefitId','BenefitName','Provider'

まとめ

サブスクリプションの請求データを取得して、Azure Blob Storage に Export することを確認しました。