Examples Section

Retrieve product information

The following example authenticates against the Open Publishing API and retrieves all products:

#!python3
import requests

API_HOST = "api.openpublishing.com"
API_KEY = "XXXXXXX"


def obtain_bearer_token(api_host, api_key):
    """Authenticate with api_key."""
    auth_url = "https://{}/auth/auth".format(api_host)
    res = requests.post(
        auth_url, data={"type": "api_key", "api_key": api_key}, timeout=5
    )
    res.raise_for_status()
    res_json = res.json()
    return res_json["auth_token"]


def get_all_products(api_host, api_key):
    """Retrieve all products from API."""
    auth_token = obtain_bearer_token(api_host, api_key)
    headers = {
        "Authorization": f"Bearer {auth_token}",
        "Content-type": "application/json",
        "Accept": "text/plain",
    }
    url = f"https://{api_host}/resource/v3/products"
    res = requests.get(url, headers=headers)
    res.raise_for_status()
    return res.json()


if __name__ == '__main__':
    all_products = get_all_products(API_HOST, API_KEY)
    print(all_products)

Listen to sales

The following example listens for new sales via the sales report API. It can be used to synchronise sales with an external reporting system:

import time
import logging
import requests


API_HOST = "api.openpublishing.com"
API_KEY = "XXXXXXX"


def obtain_bearer_token(api_host, api_key):
    """Authenticate with api_key."""
    auth_url = "https://{}/auth/auth".format(api_host)
    res = requests.post(
        auth_url, data={"type": "api_key", "api_key": api_key}, timeout=5
    )
    res.raise_for_status()
    res_json = res.json()
    return res_json["auth_token"]


def request_next_sales(api_host, bearer_token, from_sale_id: int):
    """Request next 500 sales from sales resource."""
    url = f"https://{api_host}/resource/v3/sales?id__gt={from_sale_id}&display=500"
    headers = {"Authorization": f"Bearer {bearer_token}"}
    res = requests.get(url, headers=headers)
    res.raise_for_status()
    data = res.json()
    return [sale for _, sale in data.items()]


def handle_sales(sales):
    """Do something useful with sales."""
    for sale in sales:
        logging.warning("%s: %s %s %.2f %s ", sale['period'], sale['ean'], sale['sales_channel'],
            sale['amount_received']/100, sale['amount_received_currency'])


def fetch_sales_forever(api_host, bearer_token, from_sale_id = None):
    """Fetch all sales in an endless loop, sleep if no new sales present."""
    while True:
        sales = request_next_sales(api_host, bearer_token, from_sale_id)
        sales = sorted(sales, key=lambda s: s["id"])
        handle_sales(sales)
        if sales:
            from_sale_id = sales[-1]["_id"]
        else:
            # No new sales.
            time.sleep(15)


if __name__ == '__main__':
    bearer_token = obtain_bearer_token(API_HOST, API_KEY)
    fetch_sales_forever(API_HOST, bearer_token)