Vitals API V2リファレンス

import time

import requests

# 指定しなければならない値
client_id = "<クライアントID>"
client_secret = "<クライアントシークレット>"
package_name = "<アプリパッケージ名>"

BASE_URL = "https://developer.amazon.com/api/appstore"
VITALS_URL = f"{BASE_URL}/v2/applications/{package_name}/vitals"

_token = {"value": None, "expires_at": 0.0}


def auth_headers():
    """Return the request headers with a valid LWA access token."""
    if time.time() > _token["expires_at"] - 300:
        response = requests.post(
            "https://api.amazon.com/auth/o2/token",
            data={
                "grant_type": "client_credentials",
                "client_id": client_id,
                "client_secret": client_secret,
                "scope": "adx_reporting::appstore:marketer",
            },
            timeout=30,
        )
        response.raise_for_status()
        token = response.json()
        _token["value"] = token["access_token"]
        _token["expires_at"] = time.time() + token["expires_in"]
    return {
        "Authorization": f"Bearer {_token['value']}",
        "Content-Type": "application/json",
        "Accept": "application/json",
    }

import random
import time

import requests

# 「アクセストークンを取得」からauth_headers()と_tokenを使用します。

MAX_ATTEMPTS = 4
RETRYABLE_STATUS_CODES = {429, 500, 503, 504}


def send(method, url, body=None):
    """Send a request.Retry a throttled request or a temporary error."""
    token_renewed = False
    for attempt in range(1, MAX_ATTEMPTS + 1):
        headers = auth_headers()
        try:
            response = requests.request(
                method, url, headers=headers, json=body, timeout=60
            )
        except (requests.ConnectionError, requests.Timeout):
            if attempt == MAX_ATTEMPTS:
                raise
            time.sleep(2 ** attempt + random.uniform(0, 1))
            continue

        renew = response.status_code == 401 and not token_renewed
        if renew and attempt < MAX_ATTEMPTS:
            token_renewed = True
            _token["expires_at"] = 0.0  # 次回の試行に備えて、新しいトークンを取得してください。
            continue
        if response.status_code not in RETRYABLE_STATUS_CODES:
            response.raise_for_status()
            return response.json()
        if attempt == MAX_ATTEMPTS:
            response.raise_for_status()

        retry_after = response.headers.get("Retry-After")
        if retry_after is not None:
            delay = float(retry_after)
        else:
            delay = 2 ** attempt + random.uniform(0, 1)
        time.sleep(delay)

from datetime import datetime, timedelta

TIME_FORMAT = "%Y-%m-%dT%H:%M:%SZ"


def time_range(metric_set_name, aggregation_period, periods):
    """Return (startTime, endTime) for the latest time periods of a metric set."""
    info = send("GET", f"{VITALS_URL}/{metric_set_name}")
    latest = {item["aggregationPeriod"]: item["latestDataTimestamp"] for item in info["freshnessInfo"]}
    end = datetime.strptime(latest[aggregation_period], TIME_FORMAT)
    step = timedelta(days=1) if aggregation_period == "DAILY" else timedelta(hours=1)
    start = end - step * (periods - 1)
    return start.strftime(TIME_FORMAT), end.strftime(TIME_FORMAT)

start_time, end_time = time_range("crashMetricSet", "DAILY", 7)
crash_body = {
    "timelineSpec": {"aggregationPeriod": "DAILY", "startTime": start_time, "endTime": end_time},
    "metrics": ["crashRate", "crashCount", "distinctDevices"],
    "dimensions": ["versionCode"],
    "filter": {"countryCode": ["US", "CA"], "deviceType": ["FIRE_TV"]},
}

result = send("POST", f"{VITALS_URL}/crashMetricSet:query", crash_body)
for row in result["rows"]:
    values = {metric["name"]: metric["value"] for metric in row["metrics"]}
    print(row["startTime"], row["dimensions"], values)

start_time, end_time = time_range("appLatencyMetricSet", "DAILY", 7)
result = send("POST", f"{VITALS_URL}/appLatencyMetricSet:query", {
    "timelineSpec": {"aggregationPeriod": "DAILY", "startTime": start_time, "endTime": end_time},
    "metrics": ["coldStartAppLaunchTime", "warmStartAppLaunchTime"],
    "dimensions": ["deviceType"],
})

start_time, end_time = time_range("slowStartRateMetricSet", "HOURLY", 24)
slow_start_body = {
    "timelineSpec": {"aggregationPeriod": "HOURLY", "startTime": start_time, "endTime": end_time},
    "metrics": ["slowStartRate", "slowStartRate7dDeviceWeighted", "distinctDevices"],
    "dimensions": ["startType"],
    "filter": {"startType": ["COLD"]},
}

result = send("POST", f"{VITALS_URL}/slowStartRateMetricSet:query", slow_start_body)
for row in result["rows"]:
    values = {metric["name"]: metric["value"] for metric in row["metrics"]}
    print(row["startTime"], values["slowStartRate"], values["distinctDevices"])

from datetime import timezone

try:
    start_time, end_time = time_range("readyToUseMetricSet", "DAILY", 7)
except KeyError:
    # まだ更新されていません。最新の UTC 日付を照会して可用性の状態を取得してください。
    end = datetime.now(timezone.utc)
    start_time = (end - timedelta(days=6)).strftime(TIME_FORMAT)
    end_time = end.strftime(TIME_FORMAT)

ready_to_use_body = {
    "timelineSpec": {"aggregationPeriod": "DAILY", "startTime": start_time, "endTime": end_time},
    "metrics": ["coldStartReadyToUseTime", "warmStartReadyToUseTime"],
}

result = send("POST", f"{VITALS_URL}/readyToUseMetricSet:query", ready_to_use_body)
availability = result["availability"]
if availability["state"] == "ACTION_REQUIRED":
    print(availability["message"])
    print(availability["documentationUrl"])

for row in result["rows"]:
    for metric in row["metrics"]:
        print(row["startTime"], metric["name"], metric["statistics"]["Average"])

def query_all_rows(url, body):
    """Send a query.Return the rows of all pages and the freshness of the last page."""
    rows = []
    page_body = dict(body)
    page_body.pop("pageToken", None)
    while True:
        result = send("POST", url, page_body)
        rows.extend(result["rows"])
        if result["nextPageToken"] is None:
            return rows, result.get("freshness")
        page_body["pageToken"] = result["nextPageToken"]
        time.sleep(0.4)

start_time, end_time = time_range("crashMetricSet", "DAILY", 14)
issues_body = {
    "timelineSpec": {"aggregationPeriod": "DAILY", "startTime": start_time, "endTime": end_time},
    "reportType": "CRASH",
    "filter": {"versionCode": ["152"]},
    "pageSize": 25,
}

result = send("POST", f"{VITALS_URL}/issuesMetricSet:query", issues_body)
for issue in result["rows"]:
    counts = {metric["name"]: metric["value"] for metric in issue["metrics"]}
    print(issue["crashDescriptor"], counts["errorEventCount"], counts["affectedDeviceCount"])
    print(issue["reportText"])