aaapi.py

from __future__ import annotations

import functools
import gzip
import json
from time import sleep, time as now
from typing import List

import requests


class AdvertisingAPI(object):
    """Client Library"""

    def __init__(self, client_id: str, client_secret: str, refresh_token: str) -> AdvertisingAPI:
        """

            :param client_id: Whitelisted client id (LWA) with grants for the scope `cpc_advertising:campaign_management`
            :param client_secret: LWA (Login With Amazon) Client secret
            :param refresh_token: Refresh token for the Advertiser Account

        """

        self.client_id = client_id
        self.client_secret = client_secret
        self.refresh_token = refresh_token

        # Access token for the Advertiser Account
        self._access_token = dict(
            token=None,
            expiration=None,
        )

        # Profiles identify a specific Advertiser in a given Marketplace
        self._profile_id = None

        self._update_access_token()

        # Specific headers required in every call
        self._auth_headers = {
            'Content-Type': 'application/json',
            'Authorization': f'bearer {self._access_token["token"]}',
            'Amazon-Advertising-API-ClientId': self.client_id,
        }

    def _set_profile_id(self, profile_id):
        """Set Profile ID basically set the API Scope in the header's request"""

        self._auth_headers['Amazon-Advertising-API-Scope'] = str(profile_id)

    def _update_access_token(self):

        payload = {
            'grant_type': 'refresh_token',
            'client_id': self.client_id,
            'refresh_token': self.refresh_token,
            'client_secret': self.client_secret,
        }
        dat = requests.post('https://api.amazon.com/auth/o2/token', data=payload).json()
        self._access_token['token'] = dat['access_token']
        self._access_token['expiration'] = now() + 3600

    def _check_token_expiration(func):
        @functools.wraps(func)
        def wrap(self, *args, **kwargs):
            expiration = self._access_token['expiration']
            if expiration:
                if expiration < now():
                    self._update_access_token()
            return func(self, *args, **kwargs)

        return wrap

    @_check_token_expiration
    def get_profiles(self) -> List[dict]:
        """Retrieves profiles associated with an auth token"""

        return requests.get('https://advertising-api-eu.amazon.com/v2/profiles', headers=self._auth_headers).json()

    @_check_token_expiration
    def fetch_object(self, url: str, gunzip: bool = False) -> List[dict]:
        """Utility method to fetch reports or snapshots already wrapped as Python objects"""
        print(f"fetching report from url: {url}")
        r = requests.get(url, headers=self._auth_headers, stream=True)
        if gunzip:
            return json.loads(gzip.decompress(r.content))
        return r.json()

    @_check_token_expiration
    def fetch_sb_report(self,
                     profile_id: int,
                     record_type: str,
                     report_date: str,
                     metrics: List[str],
                     waiting_time: int = 1,
                     campaign_type: str = 'sponsoredProducts') -> List[dict]:

        self._set_profile_id(profile_id)

        metrics_as_string = ",".join(metrics)
        req_url = f'https://advertising-api-eu.amazon.com/v2/hsa/{record_type}/report'
        campaign_type_data = ''
        if record_type == 'asins':
            campaign_type_data = f'"campaignType": "{campaign_type}",'

        data = """{{"reportDate": {}, {} "metrics": "{}" }}""".format(report_date, campaign_type_data,
                                                                      metrics_as_string)

        r = requests.post(req_url, data=data, headers=self._auth_headers).json()
        report_id = r.get("reportId")

        if not report_id:
            raise Exception(f'No report available: {r}')

        print(f"report requested, got id:{report_id}")

        check_url = f'https://advertising-api-eu.amazon.com/v2/reports/{report_id}'
        attempts = 1

        while True:
            try:
                print(f"checking report generation status from: {check_url}")
                response = requests.get(check_url, headers=self._auth_headers).json()
                url = response['location']
                print(f"Successful response: {response}")
                break
            except KeyError as e:
                if response.get('status') != 'IN_PROGRESS':
                    raise Exception(f'Error downloading report: {response}')
                print(f"Report status: {response.get('status')}")
                sleep(waiting_time)
                attempts += 1
        print(f"downloading report from: {url}")
        return self.fetch_object(url, gunzip=True)

main.py

import os
import json

import dotenv

from aaapi import AdvertisingAPI

dotenv.load_dotenv()

api = AdvertisingAPI(
    client_id="",
    client_secret="",
    refresh_token="",
)

profiles = api.get_profiles()

try:
    profile = profiles[0]["profileId"]
except KeyError as e:
    print(f"no profiles found: {e} ")

record_type = "campaigns"
url = f"https://advertising-api-eu.amazon.com/v2/hsa/{record_type}/report"

metrics = [
    "campaignName",
    "campaignId",
    "campaignStatus",
    "campaignBudget",
    "campaignBudgetType",
    "campaignRuleBasedBudget",
    "applicableBudgetRuleId",
    "applicableBudgetRuleName",
    "attributedConversions14d",
    "attributedConversions14dSameSKU",
    "attributedConversions14dSameSKU",
    "attributedUnitsOrdered14d",
    "attributedSales14d",
    "attributedSales14dSameSKU",
]
metrics_as_string = ",".join(metrics)
report_date = "20210401"

obj = api.fetch_sb_report(profile, record_type, report_date, metrics, waiting_time=25)

with open("report.cvs", "w") as report_file:
    json.dump(obj, report_file, sort_keys=True, indent=4, ensure_ascii=False)