I’m looking to access Looks, Slugs, etc. via the Python API from a Jupyter Notebook as advertised in the “Looker for Data Science” marketing materials:
I’ve tried a couple Python API versions, but both fetch Look functions I’ve tried returned a CSV truncated to 500 rows. Here’s some code I’ve tried, is anyone able to show me how to modify it to return the full dataset, or post some code that’d achieve that goal?:
import requests, json, logging, time, urllib
import pandas as pd
from io import StringIO
import io as stringIOModule
from datetime import datetime
import yaml
logger = logging.getLogger(__name__)
logger.setLevel(logging.INFO)
handler = logging.FileHandler(str(time.strftime("%d_%m_%Y")) +"_looker_API_Calls" + ".log")
handler.setLevel(logging.INFO)
formatter = logging.Formatter('%(asctime)s - %(name)s - %(levelname)s - %(message)s')
handler.setFormatter(formatter)
logger.addHandler(handler)
logger.info('Testing Logs')
class LookerAPI(object):
"""Class that contains methods and variables related to looker API authentication"""
def __init__(self, api_info):
self.api_endpoint = api_info['api_endpoint']
self.client_secret = api_info['client_secret']
self.client_id = api_info['client_id']
self.login_endpoint = api_info['login_url']
print(self.login_endpoint)
def login(self):
"login to looker API"
try:
auth_data = {'client_id':self.client_id, 'client_secret':self.client_secret}
r = requests.post( self.login_endpoint,data=auth_data) # error handle here
json_auth = json.loads(r.text)['access_token']
return json_auth
except requests.exceptions.RequestException as e:
logger.error(e)
def run_look(self, look_id, json_auth,return_format='csv'):
"run looks"
try:
look_run_url = self.api_endpoint + '/looks/{0}/run/{1}'.format(look_id,return_format)
#r = requests.get(look_run_url, headers={'Authorization': "token " + json_auth})
r = requests.get(look_run_url + '?' + 'access_token=' + json_auth)
return r.text
except requests.exceptions.RequestException as e:
logger.error(e)
def run_query(self, query_id, json_auth, return_format='csv'):
"run query"
try:
query_run_url = self.api_endpoint + '/queries/{0}/run/{1}'.format(query_id,return_format)
#r = requests.get(query_run_url, headers={'Authorization': "token " + json_auth})
r = requests.get(query_run_url + '?' + 'access_token=' + json_auth)
return r.text
except requests.exceptions.RequestException as e:
logger.error(e)
def run_query_slug(self, query_slug, json_auth):
"run query"
try:
query_slug_run_url = self.api_endpoint + '/queries/slug/{0}'.format(query_slug)
#r = requests.get(query_run_url, headers={'Authorization': "token " + json_auth})
r = requests.get(query_slug_run_url + '?' + 'access_token=' + json_auth)
qid=json.loads(r.text)["id"]
print("Query_id: " + str(qid))
return LookerAPI.run_query(self, qid, json_auth)
except requests.exceptions.RequestException as e:
logger.error(e)
host = 'localhost'
f = open('config.yml')
params = yaml.load(f, Loader=yaml.FullLoader)
f.close()
creds = {
'api_endpoint' : params['hosts'][host]['api_endpoint'],
'login_url': params['hosts'][host]['login_url'],
'client_id' : params['hosts'][host]['my_token'],
'client_secret': params['hosts'][host]['my_secret']
}
looker = LookerAPI(creds)
json_auth = looker.login()
look = looker.run_look(1633, json_auth)
records = pd.read_csv(StringIO(look))
print(records.shape)
records.head(5)
Result -> (500, 42)
