[CI-DWH][Python] Upload CSV To Be Processed by the Post ETL Script for the CI DWH

The following script will upload a CSV to NetApp Cloud Insights, so that it can be processed by the Post ETL script, in order to update the Cloud Insights Data Warehouse dwh_custom.

Save as say ci_upload_csv.py and run as:

python ci_upload_csv.py TENANT_URL API_TOKEN CSV_FILE

Note 1: The following script doesn't include code for being behind a proxy, if you need this then hints are here: Using Python with Proxy and Authentication. Or you could enter proxy details into the Command Prompt like:


Note 2: This is based on the NetApp CI module, just with the upload to DataWarehouse bit extracted and turned into a simplified, standalone tool. Also added verify=False in there (since where I'm working the SSL cert gets intercepted and modified.)

The Script

#!/usr/bin/env python

''' Upload csv file to be processed by Data Warehouse custom script '''

import argparse
from pathlib import Path
from datetime import datetime
from enum import Enum
import requests
import urllib3

''' START OF CLASS Client '''

class Client:

  def __init__(self, instance, token, version='v1', session=None):
    self.host = instance.rstrip('/') # Trim trailing slash
    self.version = version
    self.base_path = f'/rest/{version}'
    if session is None:
      session = requests.Session()
      'Content-Type': 'application/json',
      'X-CloudInsights-ApiKey': token,
    self.session = session
  def __enter__(self):
    return self
  def __exit__(self, type, value, traceback):
    return False
  def _convert_to_fully_qualified_url(self, api_path):
    if api_path.startswith(self.host):
      # Fully qualified URL
      return api_path
    if api_path.startswith(self.base_path):
      # URL that is only missing the schema and server
      return self.host + api_path
    # The URL is relative - needs to add prefix
    return f'{self.host}{self.base_path}/{api_path.lstrip("/")}'
  def _request(self, method, path, **kwargs):
    return_response = kwargs.pop('return_response', None)
    response = self.session.request(
    if return_response:
      return response
    # Certain calls like /dwh-management/upload/csvs return an empty result
    if not response.text:
      return {}
    return response.json()
  def post(self, path, json=None, **kwargs):
    return self._request('POST', path, json=json, **kwargs)
  def _raise_for_status(response):
    ''' Improve the original raise_for_status in the requests library to add some logging '''
    if 400 <= response.status_code < 500:
      # For client error, raise a regular HTTP error
      raise requests.HTTPError(f'{response.status_code} Client Error: {response.reason} {response.text}', response=response)
    elif 500 <= response.status_code < 600:
      # For server-side errors, we need further parsing to extract OCI specific error information
      raise APIError(f'{response.status_code} Server Error: {response.reason}', response=response)
  def upload_dwh_csv(self, filename):
    ''' Upload a CSV file to be processed by Data Warehouse. '''
    return self.post(
      files={'customFile': open(filename, 'rb')},
      headers={'Content-Type': None},  # Reset Content-Type from default
      return_response=True  # Get the response object, as no JSON is returned for this call
  def close(self):
    ''' Close the session. '''
    self.session = None

''' END OF CLASS Client '''


def upload_file(client, csv_file):
  if not Path(csv_file).is_file():
    print(f'File "{csv_file}" does not exist or is not a file.')
    response = client.upload_dwh_csv(csv_file)
    if response.ok: print('File uploaded successfully!')
    else: print(f'Error posting file: {response.text}')
  except Exception as e: print(e)

if __name__ == '__main__':
  parser = argparse.ArgumentParser()
  parser.add_argument('instance', help='Cloud Insights Instance')
  parser.add_argument('token', help='Cloud Insights Token')
  parser.add_argument('csv', help='CSV upload file')
  options = parser.parse_args()
  with Client(options.instance, options.token) as client:
    upload_file(client, options.csv)