211 lines
7.1 KiB
Python
211 lines
7.1 KiB
Python
import sys
|
|
import base64
|
|
import requests
|
|
import datetime
|
|
import json
|
|
|
|
# added this to silence annoying SSL warnings in requests
|
|
import urllib3
|
|
urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)
|
|
|
|
class AFCBusiness:
|
|
def __init__( self, api_url=None, api_jwt=None, username=None, password=None ):
|
|
# Update this as we modify the library for newer versions
|
|
self.api_version = 'v1'
|
|
|
|
self.api_jwt = api_jwt
|
|
#self.api_url = f"https://afaccount.roosevelt.hyperscaleaf.com/api/{self.api_version}/businesses"
|
|
|
|
# Dev
|
|
self.api_url = f"https://afaccount.eisenhower.hyperscaleaf.com/api/"
|
|
|
|
# Prod
|
|
#self.api_url = f"https://afaccount.roosevelt.hyperscaleaf.com/api/"
|
|
|
|
self.credentials = {}
|
|
self.credentials['user_name'] = username #'rross'
|
|
self.credentials['password'] = password #'rGstj!f672q5FBZ'
|
|
|
|
self.headers = {}
|
|
self.headers['content-type'] = "application/json"
|
|
self.headers['accept'] = "application/json"
|
|
|
|
if self.api_jwt == None:
|
|
if self.credentials['user_name'] and self.credentials['password']:
|
|
self.Set_JWT()
|
|
else:
|
|
print("Unable to connect to API without jwt or authentication credentials")
|
|
sys.exit(1)
|
|
|
|
self.headers['authorization'] = "Bearer " + self.api_jwt
|
|
|
|
self.params = {}
|
|
|
|
def Set_JWT( self ):
|
|
url = f"{self.api_url}{self.api_version}/auth/authenticate"
|
|
|
|
try:
|
|
response = requests.post(url, headers=self.headers, json=self.credentials)
|
|
except requests.exceptions.RequestException as e:
|
|
print("exception: {}".format(e))
|
|
sys.exit(1)
|
|
|
|
try:
|
|
response.raise_for_status()
|
|
except requests.exceptions.HTTPError as err:
|
|
print("http error: {}".format(err))
|
|
sys.exit(1)
|
|
|
|
rj = response.json();
|
|
self.api_jwt = rj['jwt']
|
|
|
|
return
|
|
|
|
def Submit_Get( self, full_url, input_params={} ):
|
|
if not isinstance(full_url, str):
|
|
print("Submit_Get() error: must provide the full url string")
|
|
return None
|
|
|
|
try:
|
|
self.response = requests.get(full_url, verify=False, headers=self.headers, params=input_params)
|
|
#print("debug: url = {}".format(self.response.url))
|
|
except requests.exceptions.RequestException as e:
|
|
print("exception: {}".format(e))
|
|
#sys.exit(1)
|
|
return None
|
|
|
|
try:
|
|
self.response.raise_for_status()
|
|
except requests.exceptions.HTTPError as err:
|
|
print("http error: {}".format(err))
|
|
#sys.exit(1)
|
|
return None
|
|
|
|
return self.response
|
|
|
|
def Submit_Post( self, full_url, input_params={}, data={} ):
|
|
if not isinstance(full_url, str):
|
|
print("Submit_Post() error: must provide the full url string")
|
|
return None
|
|
|
|
print(f"\ndata = {data}")
|
|
|
|
try:
|
|
this_headers = self.headers
|
|
this_headers['Content-Type'] = 'application/json'
|
|
#self.response = requests.post(full_url, verify=False, headers=this_headers, params=input_params, data=data)
|
|
self.response = requests.post(full_url, verify=False, headers=this_headers, data=json.dumps(data))
|
|
#print("debug: url = {}".format(self.response.url))
|
|
except requests.exceptions.RequestException as e:
|
|
print("exception: {}".format(e))
|
|
#sys.exit(1)
|
|
return None
|
|
|
|
try:
|
|
self.response.raise_for_status()
|
|
except requests.exceptions.HTTPError as err:
|
|
print("http error: {}".format(err))
|
|
print(f"\nself.response.attributes = {self.response.__dict__}")
|
|
#sys.exit(1)
|
|
return None
|
|
|
|
return self.response
|
|
|
|
def Submit_Patch( self, full_url, input_params={}, data={} ):
|
|
if not isinstance(full_url, str):
|
|
print("Submit_Patch() error: must provide the full url string")
|
|
return None
|
|
|
|
#print(f"\ndata = {data}")
|
|
|
|
try:
|
|
this_headers = self.headers
|
|
this_headers['Content-Type'] = 'application/json'
|
|
#self.response = requests.post(full_url, verify=False, headers=this_headers, params=input_params, data=data)
|
|
self.response = requests.patch(full_url, verify=False, headers=this_headers, data=json.dumps(data))
|
|
#print("debug: url = {}".format(self.response.url))
|
|
except requests.exceptions.RequestException as e:
|
|
print("exception: {}".format(e))
|
|
#sys.exit(1)
|
|
return None
|
|
|
|
try:
|
|
self.response.raise_for_status()
|
|
except requests.exceptions.HTTPError as err:
|
|
print("http error: {}".format(err))
|
|
print(f"\nself.response.attributes = {self.response.__dict__}")
|
|
#sys.exit(1)
|
|
return None
|
|
|
|
return self.response
|
|
|
|
def GetBusinesses(self):
|
|
business_url = f"{self.api_url}{self.api_version}/businesses"
|
|
input_params = {}
|
|
|
|
# offset number in list
|
|
#input_params['start'] = 0
|
|
|
|
# max number of businesses to return in list
|
|
input_params['limit'] = 200
|
|
|
|
# filter to these status values
|
|
#input_params['status'] = ['active','archived']
|
|
|
|
r = self.Submit_Get(business_url, input_params)
|
|
results = r.json()
|
|
|
|
if 'values' in results:
|
|
# print(f"AFC Business List")
|
|
# for b in results['values']:
|
|
# print(f"{b}\n")
|
|
return results['values']
|
|
|
|
return None
|
|
|
|
def PostBusiness(self, business=None):
|
|
business_url = f"{self.api_url}{self.api_version}/businesses"
|
|
input_params = {}
|
|
|
|
# business_object = {
|
|
# 'name' : 'string',
|
|
# "account_id_legacy": "string",
|
|
# "primary_contact_name": "string",
|
|
# "primary_contact_phone": "string",
|
|
# "primary_contact_email": "string",
|
|
# "company_public_name": "string",
|
|
# "logo_url": "string",
|
|
# "bandwidth_campaign_id": "string",
|
|
# "reports_required": true,
|
|
# "sms_base_price_contract": 0,
|
|
# "mms_base_price_contract": 0,
|
|
# "mms_included_characters": 0,
|
|
# "afc_connect_base_price_contract": 0,
|
|
# "afc_crystal_pricing": 0
|
|
# }
|
|
|
|
r = self.Submit_Post(business_url, input_params, data=business)
|
|
if r == None:
|
|
return None
|
|
|
|
results = r.json()
|
|
return results
|
|
|
|
def PatchBusiness(self, business_key=None, fields=None):
|
|
if business_key == None:
|
|
print(f"Can't patch a business without the business key")
|
|
return None
|
|
|
|
if fields == None:
|
|
print(f"Can't patch a business without a dict of fields")
|
|
return None
|
|
|
|
business_url = f"{self.api_url}{self.api_version}/businesses/{business_key}"
|
|
|
|
r = self.Submit_Patch(business_url, input_params={}, data=fields)
|
|
if r == None:
|
|
return None
|
|
|
|
results = r.json()
|
|
return results
|