Pull JSON Data From Endpoint and Flatten it to CSV

Posted on Nov 18, 2020

Sometimes as a QA Engineer,you need to pull test data from endpoints from third parties.That is clients you are integrating you system with.

This is usually vital when the developers are working on the integration. You can start preparing the tests early enough.

Example script shows how to pull credit score data in json and flatten it to csv.

Better Example Now Using Pandas

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
import pandas as pd
from requests.auth import HTTPBasicAuth
import requests
import json
from pandas.io.json import json_normalize

# import specific column
# measure time taken to import big CSV
df = pd.read_csv('farmers_identification.csv',usecols=['farmer_national_id'],verbose=True)

# print(df)
token = ''

def get_auth_token(token):
  endpoint='https://test-gateway.tulaa.io/uaa-server/oauth/token'

  payload = {
    'username':'<my-email>',
    'password':'<my-password>',
    'grant_type':'password'
  }

  response = requests.post(endpoint,auth=HTTPBasicAuth('<web_client>','<client_secret>'),data=payload)

  if(response.ok):

    jData = json.loads(response.content)

    for key in jData:
      if key == 'access_token' in jData:
        token='Bearer'+' '+str(jData[key])
        # print(token)

      return token
  else
    # my_logger.debug('a debug message')
    raise ApiError('Cannot fetch access token:{}'.format(response.status_code))

def get_credit_score(row):

  token = get_auth_token(token)

  headers ={
    'Authorization':token
  }

  try:
    url = 'https://test-cogency-service.tulaa.io/credit/score/str[row]'

    response = (requests.get(url).text)
    response_json = json.loads(response)

    return response_json


  except Exception as e:
    raise e


  df['API_response'] = df.apply(get_credit_score,axis=1)
  df['API_response'].head()

  new_df = json_normalize(df['API_response'])
  new_df = new_df[['metropol_credit_score','id_number']]
  new_df

  new_df.to_csv(path_or_buf='./farmers-with-score.csv',index=False)


if __name__ == '__main__':
  get_credit_score()