catalog.py

import argparse
import requests
import logging as log 
import pandas as pd
from datetime import datetime
import json

def call_exponea_api(method,path, payload, config):
    url = f"{config['api_endpoint']}{path}"
    log.debug(url)

    headers = {
        "accept" : "application/json",
        "authorization" : f"Basic {config['token']}"
    }
    if method == "POST":
        response = requests.post(
            url = url,
            json= payload,
            headers=headers
        )
    elif method == "GET":
        response = requests.get(
            url = url,
            headers=headers
        )

    log.debug(response.status_code)
    log.debug(response.content)

    data = response.json()
    return data


parser = argparse.ArgumentParser(
                    prog='bloom-catalog.py',
                    description='Manipulate with EXPONEA catalog API',
                    epilog='@author: [email protected]')

parser.add_argument('command', 
                    choices=["export","list","testimport"],
                    help='command. Actually only "export" is supported')

parser.add_argument('--catalog-id',help = "catalog id", required=True )
parser.add_argument('--file',help = "output filename", required=True )
parser.add_argument('--api-endpoint',help = "api andpoint URL", required=True)
parser.add_argument('--project-token',help = "project token", required=True)
parser.add_argument('--api-token', help="private token", required=True)
                     

args = parser.parse_args()
log.basicConfig(level=log.INFO)

config = {
    "api_endpoint": args.api_endpoint,
    "token": args.api_token
}


if args.command == "export":
    #TODO: add pagination
    log.info(f"Exporting catalog {args.catalog_id} to file")
    data = call_exponea_api(
        "GET",
        f"/data/v2/projects/{args.project_token}/catalogs/{args.catalog_id}/items?count=1000",
        {},
        config
    )

    def modify_catalog_item(item):
        out = item["properties"]
        out["item_id"] = item["item_id"]
        return out

    data = list(map(modify_catalog_item, list(data["data"])))

    df = pd.read_json(json.dumps(data))

    filename = args.file
    df.to_csv(
        filename, index=False
    )

    log.info(f"{len(data)} rows exported to {filename}")
elif args.command == "testimport":
    data = {
        "properties":{
            "attr1":"imported by python"
        },
        "upsert":True
    }
    log.info(f"Importing data to catalog {args.catalog_id} ")
    data = call_exponea_api(
        "POST",
        f"/data/v2/projects/{args.project_token}/catalogs/{args.catalog_id}/items/id1/partial-update",
        data,
        config
    )
else:
    log.info("unknown command f{args.catalog_id}")