Domo Connector for Python

domo.py

from http.client import HTTPSConnection
from base64 import b64encode
import json
import pandas

class DomoConnector:
  def __init__(self, client_id, client_secret):
    self.__client_id = client_id
    self.__client_secret = client_secret
  
  def __basic_auth(self, username, password):
    token = b64encode(f"{username}:{password}".encode('utf-8')).decode("ascii")
    return f'Basic {token}'

  def __get_oauth_token(self):
    c = HTTPSConnection("api.domo.com")
    headers = { 'Authorization' : self.__basic_auth(self.__client_id, self.__client_secret) }
    c.request('GET', '/oauth/token?grant_type=client_credentials&scope=data', headers=headers)
    res = c.getresponse()
    data = res.read()
    test = json.loads(data)  
    return test['access_token']

  def update(self, data_set_id, json_data, append = False):
    access_token = self.__get_oauth_token();
    conn = HTTPSConnection("api.domo.com")
    payload = json.dumps(json_data)
    headers = {
      'Authorization': 'Bearer ' + access_token,
      'Content-Type': 'application/json'
    }
    conn.request("PUT", "/v1/json/" + data_set_id + "/data" + ('?updateMethod=APPEND' if append else ''), payload, headers)
    res = conn.getresponse()
    data = res.read()
    return data.decode("utf-8")
  
  def append(self, data_set_id, json_data):
    return self.update(data_set_id, json_data, True)

  def query(self, data_set_id, query_str = 'SELECT * FROM table'):
    access_token = self.__get_oauth_token();
    conn = HTTPSConnection("api.domo.com")
    payload = json.dumps({"sql": query_str})
    headers = {
      'Authorization': 'Bearer ' + access_token,
      'Content-Type': 'application/json'
    }
    conn.request("POST", "/v1/datasets/query/execute/" + data_set_id, payload, headers)
    res = conn.getresponse()
    data = res.read()
    _json = json.loads(data.decode("utf-8"))
    col_headers = _json['columns']
    data_rows = _json['rows']
    df = pandas.DataFrame(data=data_rows, columns=col_headers)
    return df