This commit is contained in:
Zack Meier
2026-04-15 15:45:50 -05:00
commit 1d304511b8
613 changed files with 140998 additions and 0 deletions
@@ -0,0 +1,502 @@
import requests
import json
import os
import pytz
from datetime import datetime
from datetime import timedelta
from datetime import time
# API Documentation
# https://docs.servicenow.com/bundle/sandiego-application-development/page/build/applications/concept/api-rest.html
class SnowAPI:
def __init__(self, snInstance):
if not ((os.environ.get('SN_USER')) and (os.environ.get('SN_PASS'))):
print("\n*** Environment variables are not set for ServiceNow Authentication ***")
exit()
self.snInstance = snInstance
self.headers = {
'Content-Type': 'application/json',
'Accept': 'application/json'
}
self.user = os.environ['SN_USER']
self.pwd = os.environ['SN_PASS']
def getGroupID(self, grpName):
grpName = grpName.replace(" ","%20")
url = 'https://{}/api/now/table/sys_user_group?sysparm_query=name={}'.format(self.snInstance, grpName)
apiResponse = requests.get(url, auth=(self.user, self.pwd), headers=self.headers)
apiData = apiResponse.json()
return apiData['result'][0]['sys_id']
def getUserID(self, userName):
userName = userName.replace(" ","%20")
apiCall = 'https://{}/api/now/table/sys_user?sysparm_query=user_name={}'.format(self.snInstance, userName)
apiResponse = requests.get(apiCall, auth=(self.user, self.pwd), headers=self.headers)
apiData = apiResponse.json()
return apiData['result'][0]['sys_id']
def getCMDBItemByIP(self, pattern):
apiCall = "https://{0}/api/now/table/cmdb_ci_server?sysparm_query=ip_address={1}&sysparm_limit=10".format(self.snInstance, pattern)
apiResponse = requests.get(apiCall, auth=(self.user, self.pwd), headers=self.headers)
apiData = apiResponse.json()
try:
appNameRec = self.getCMDBAppById(apiData['result'][0]['u_nd_application_svc']['value'])
appName = appNameRec['name']
apiData['result'][0].update({
"u_nd_application_svc": {
"link": apiData['result'][0]['u_nd_application_svc']['link'],
"value": apiData['result'][0]['u_nd_application_svc']['value'],
"name": appName
}
})
except:
apiData['result'][0].update({
"u_nd_application_svc": {
"name": "UNDEFINED"
},
"environment": "UNDEFINED"
})
return apiData['result']
def getCMDBItemByHostName(self, pattern):
apiCall = "https://{0}/api/now/table/cmdb_ci_server?sysparm_query=host_name={1}&sysparm_limit=10".format(self.snInstance, pattern)
apiResponse = requests.get(apiCall, auth=(self.user, self.pwd), headers=self.headers)
apiData = apiResponse.json()
try:
appNameRec = self.getCMDBAppById(apiData['result'][0]['u_nd_application_svc']['value'])
appName = appNameRec['name']
apiData['result'][0].update({
"u_nd_application_svc": {
"link": apiData['result'][0]['u_nd_application_svc']['link'],
"value": apiData['result'][0]['u_nd_application_svc']['value'],
"name": appName
}
})
except:
apiData['result'][0].update({
"u_nd_application_svc": {
"name": "UNDEFINED"
},
"environment": "UNDEFINED"
})
return apiData['result']
def getCMDBItemByFQDN(self, pattern):
apiCall = "https://{0}/api/now/table/cmdb_ci_server?sysparm_query=fqdn={1}&sysparm_limit=10".format(self.snInstance, pattern)
apiResponse = requests.get(apiCall, auth=(self.user, self.pwd), headers=self.headers)
apiData = apiResponse.json()
if len(apiData['result']) > 0:
try:
appNameRec = self.getCMDBAppById(apiData['result'][0]['u_nd_application_svc']['value'])
appName = appNameRec['name']
apiData['result'][0].update({
"u_nd_application_svc": {
"link": apiData['result'][0]['u_nd_application_svc']['link'],
"value": apiData['result'][0]['u_nd_application_svc']['value'],
"name": appName
}
})
except:
print("Result 2: {0}".format(apiData))
apiData['result'][0].update({
"u_nd_application_svc": {
"name": "UNDEFINED"
},
"environment": "UNDEFINED"
})
return apiData['result']
def getCMDBAppById(self, sysId):
apiCall = "https://{0}/api/now/table/cmdb_ci_service_auto?sys_id={1}".format(self.snInstance, sysId)
apiResponse = requests.get(apiCall, auth=(self.user, self.pwd), headers=self.headers)
apiData = apiResponse.json()
return apiData['result'][0]
def getStandardChangeTemplateID(self, templateName):
# Could be improved, might throw unexpected results if the group isn't found
# Needs error handling
allData = []
apiCall = "https://{0}/api/sn_chg_rest/change/standard/template?sysparm_query=active=true".format(self.snInstance)
apiResponse = requests.get(apiCall, auth=(self.user, self.pwd), headers=self.headers)
data = apiResponse.json()
for i in data['result']:
allData.append(i)
for record in allData:
if 'sys_name' in record:
if record['sys_name']['display_value'] == templateName:
sysID = record['sys_id']['value']
break
return sysID
def getRequestItemFromReqNum(self, reqNum):
apiCall = "https://{0}/api/now/table/sc_req_item?sysparm_query=request.number={1}&sysparm_limit=1".format(self.snInstance, reqNum)
apiResponse = requests.get(apiCall, auth=(self.user, self.pwd), headers=self.headers)
data = apiResponse.json()
item = data['result'][0]['number']
return item
def getTaskNumFromReqNum(self, reqNum):
apiCall = "https://{0}/api/now/table/sc_task?sysparm_query=request.number={1}&sysparm_limit=1".format(self.snInstance, reqNum)
apiResponse = requests.get(apiCall, auth=(self.user, self.pwd), headers=self.headers)
data = apiResponse.json()
item = data['result'][0]['number']
return item
def getTaskSysIdFromReqNum(self, reqNum):
apiCall = "https://{0}/api/now/table/sc_task?sysparm_query=request.number={1}&sysparm_limit=1".format(self.snInstance, reqNum)
apiResponse = requests.get(apiCall, auth=(self.user, self.pwd), headers=self.headers)
data = apiResponse.json()
item = data['result'][0]['sys_id']
return item
def getServiceCatalogs(self):
apiCall = "https://{0}/api/sn_sc/servicecatalog/catalogs".format(self.snInstance)
apiResponse = requests.get(apiCall, auth=(self.user, self.pwd), headers=self.headers)
items = apiResponse.json()
return items
def getServiceCatalogCategories(self, sys_id):
apiCall = "https://{0}/api/sn_sc/servicecatalog/catalogs/{1}/categories".format(self.snInstance,sys_id)
apiResponse = requests.get(apiCall, auth=(self.user, self.pwd), headers=self.headers)
items = apiResponse.json()
return items
def getServiceCatalogItems(self):
apiCall = "https://{0}/api/sn_sc/servicecatalog/items?sysparm_limit=10000&sysparm_offset=0".format(self.snInstance)
apiResponse = requests.get(apiCall, auth=(self.user, self.pwd), headers=self.headers)
items = apiResponse.json()
return items
def getServiceCatalogItemByName(self, itemName):
apiCall = "https://{0}/api/sn_sc/servicecatalog/items?sysparm_limit=10000&sysparm_offset=0".format(self.snInstance)
apiResponse = requests.get(apiCall, auth=(self.user, self.pwd), headers=self.headers)
items = apiResponse.json()
for item in items['result']:
if item['name'] == itemName:
return item
def getSpecificCatalogItem(self, sysId):
apiCall = "https://{0}/api/sn_sc/servicecatalog/items/{1}".format(self.snInstance,sysId)
apiResponse = requests.get(apiCall, auth=(self.user, self.pwd), headers=self.headers)
items = apiResponse.json()
return items
def getDataFromTaskByReqNum(self, sctask, reqNum):
# This is only written to handle one task per REQ number!!!
apiCall = "https://{0}/api/now/table/sc_task?sysparm_query=request.number={1}&sysparm_fields=state%2Cnumber%2Csys_id%2Cdescription%2Cshort_description%2Cvariables.application_name%2Cvariables.additional_comments%2Cclose_notes%2Cclosed_by&sysparm_limit=1&sysparm_display_value=true".format(self.snInstance, reqNum)
apiResponse = requests.get(apiCall, auth=(self.user, self.pwd), headers=self.headers)
data = apiResponse.json()
if data['result'][0]['number'] == sctask:
items = data['result'][0]
else:
items = ""
return items
def updateTaskDescriptions(self, sysId, short_desc, desc):
apiCall = "https://{0}/api/now/table/sc_task/{1}".format(self.snInstance,sysId)
payload = {
'short_description': short_desc,
'description': desc
}
encoded_payload = json.dumps(payload)
apiResponse = requests.put(apiCall, auth=(self.user, self.pwd), headers=self.headers, data=encoded_payload)
if apiResponse.status_code != 201 and apiResponse.status_code != 200:
print('Failed to update task:')
print('\tStatus:', apiResponse.status_code)
print('\tError Response:',apiResponse.json())
exit(1)
else:
return apiResponse.json()
def assignTaskToUser(self, sysId, userName):
userID = self.getUserID(userName)
apiCall = "https://{0}/api/now/table/sc_task/{1}".format(self.snInstance,sysId)
payload = {
"assigned_to": userID
}
encoded_payload = json.dumps(payload)
apiResponse = requests.put(apiCall, auth=(self.user, self.pwd), headers=self.headers, data=encoded_payload)
if apiResponse.status_code != 201 and apiResponse.status_code != 200:
print('Failed to update task:')
print('\tStatus:', apiResponse.status_code)
print('\tError Response:',apiResponse.json())
exit(1)
else:
return apiResponse.json()
def postAppServerRequest(self,itemID,requestFor,requestBy,approver,requestType,comments,appName,env):
requestForID = self.getUserID(requestFor)
requestByID = self.getUserID(requestBy)
approvalID = self.getUserID(approver)
# Statically set these so we always force approvals to go to Computer Systems
departmentID = "f3c65cef1bfed050bba0113fad4bcb1d"
departmentCode = "112"
divisionID = "f40758231b321450bba0113fad4bcb2d"
divisionCode = "32"
payload = {
"get_portal_messages" : "true",
"sysparm_quantity" : "1",
"sysparm_no_validation" : "true",
"variables" : {
"v_approval_department" : departmentID,
"v_approval_department_code" : departmentCode,
"v_approval_division" : divisionID,
"v_approval_division_code" : divisionCode,
"v_manager" : approvalID,
"v_requested_by" : requestByID,
"v_requested_for" : requestForID,
"request_type" : requestType,
"additional_comments" : comments,
"application_name" : appName,
"environment" : env,
"require_hosting_quote" : "No",
"add_change_disaster_recovery" : "No"
}
}
apiCall = ('https://{0}/api/sn_sc/v1/servicecatalog/items/{1}/order_now').format(self.snInstance,itemID)
encoded_payload = json.dumps(payload)
apiResponse = requests.post(apiCall, auth=(self.user, self.pwd), headers=self.headers, data=encoded_payload)
if apiResponse.status_code != 201 and apiResponse.status_code != 200:
print('Failed to create ticket:')
print('\tStatus:', apiResponse.status_code)
print('\tError Response:',apiResponse.json())
exit(1)
else:
return apiResponse.json()
return apiResponse
def postGenericSerivceRequest(self,itemID,requestFor,requestBy,approver,requestType,comments):
requestForID = self.getUserID(requestFor)
requestByID = self.getUserID(requestBy)
approvalID = self.getUserID(approver)
departmentID = "f3c65cef1bfed050bba0113fad4bcb1d"
divisionID = "f40758231b321450bba0113fad4bcb2d"
payload = {
"get_portal_messages" : "true",
"sysparm_quantity" : "1",
"sysparm_no_validation" : "true",
"variables" : {
"v_approval_department" : departmentID,
"v_approval_division" : divisionID,
"v_manager" : approvalID,
"v_requested_by" : requestByID,
"v_requested_for" : requestForID,
"v_type" : requestType,
"additional_comments" : comments
}
}
apiCall = ('https://{0}/api/sn_sc/v1/servicecatalog/items/{1}/order_now').format(self.snInstance,itemID)
encoded_payload = json.dumps(payload)
apiResponse = requests.post(apiCall, auth=(self.user, self.pwd), headers=self.headers, data=encoded_payload)
if apiResponse.status_code != 201 and apiResponse.status_code != 200:
print('Failed to create ticket:')
print('\tStatus:', apiResponse.status_code)
print('\tError Response:',apiResponse.json())
exit(1)
else:
return apiResponse.json()
return apiResponse
def getGenericData(self, url):
apiCall = "https://{0}/".format(self.snInstance,url)
apiResponse = requests.get(apiCall, auth=(self.user, self.pwd), headers=self.headers)
items = apiResponse.json()
return items
def submitTicket(self, user, subject, content):
apiCall = ('https://{}/api/now/table/incident').format(self.snInstance)
userID=self.getUserID(user)
payload = {
'caller_id': userID,
'short_description': subject,
'description': content
}
encoded_payload = json.dumps(payload)
apiResponse = requests.post(apiCall, auth=(self.user, self.pwd), headers=self.headers, data=encoded_payload)
if apiResponse.status_code != 201 and apiResponse.status_code != 200:
print('Failed to create ticket:')
print('\tStatus:', apiResponse.status_code)
print('\tError Response:',apiResponse.json())
exit(1)
else:
return apiResponse.json()
def assignTicketToGroup(self, sysId, grpName):
grpId = self.getGroupID(grpName)
url = 'https://{}/api/now/table/incident/{}'.format(self.snInstance,sysId)
#Statically set the group to storage, this value was derived by decoding the web URL on storage tickets
payload = {
'assignment_group': grpId
}
encoded_payload = json.dumps(payload)
apiResponse = requests.put(url, auth=(self.user, self.pwd), headers=self.headers, data=encoded_payload)
if apiResponse.status_code != 201 and apiResponse.status_code != 200:
print('Failed to create ticket:')
print('\tStatus:', apiResponse.status_code)
print('\tError Response:',apiResponse.json())
exit(1)
else:
return apiResponse.json()
def assessNormalChange(self, sysID):
apiCall = "https://{0}/api/sn_chg_rest/change/normal/{1}".format(self.snInstance,sysID)
payload = {
"state": "assess"
}
encoded_payload=json.dumps(payload)
apiResponse = requests.patch(apiCall, auth=(self.user, self.pwd), headers=self.headers, data=encoded_payload)
return apiResponse.json()
def addStandardChangeNotes(self, sysID, txtWorkNotes):
apiCall = "https://{0}/api/sn_chg_rest/change/standard/{1}".format(self.snInstance,sysID)
payload = {
"work_notes": txtWorkNotes
}
encoded_payload=json.dumps(payload)
apiResponse = requests.patch(apiCall, auth=(self.user, self.pwd), headers=self.headers, data=encoded_payload)
return apiResponse.json()
def scheduleStandardChange(self, sysID):
apiCall = "https://{0}/api/sn_chg_rest/change/standard/{1}".format(self.snInstance,sysID)
payload = {
"state": "Scheduled"
}
encoded_payload=json.dumps(payload)
apiResponse = requests.patch(apiCall, auth=(self.user, self.pwd), headers=self.headers, data=encoded_payload)
return apiResponse.json()
def implementStandardChange(self, sysID):
apiCall = "https://{0}/api/sn_chg_rest/change/standard/{1}".format(self.snInstance, sysID)
payload = {
"state": "Implement"
}
encoded_payload=json.dumps(payload)
apiResponse = requests.patch(apiCall, auth=(self.user, self.pwd), headers=self.headers, data=encoded_payload)
return apiResponse.json()
def reviewStandardChange(self, sysID):
apiCall = "https://{0}/api/sn_chg_rest/change/standard/{1}".format(self.snInstance, sysID)
payload = {
"state": "Review"
}
encoded_payload=json.dumps(payload)
apiResponse = requests.patch(apiCall, auth=(self.user, self.pwd), headers=self.headers, data=encoded_payload)
return apiResponse.json()
def closeStandardChange(self, sysID, code, notes):
apiCall = "https://{0}/api/sn_chg_rest/change/standard/{1}".format(self.snInstance, sysID)
payload = {
"state": "Closed",
"close_code": code,
"close_notes": notes
}
encoded_payload=json.dumps(payload)
apiResponse = requests.patch(apiCall, auth=(self.user, self.pwd), headers=self.headers, data=encoded_payload)
return apiResponse.json()
def CreateNormalChange(self, grpName, assignee, coordinator, approver, category, subCategory, shortDesc, desc, justDesc, implmntPlan, riskImpact, backoutPlan, testPlan, sTime, eTime):
# Creates a normal change request that will require approval.
grpID = self.getGroupID(grpName)
userID = self.getUserID(assignee)
chngCoordID = self.getUserID(coordinator)
chngMngrID = self.getUserID(approver)
apiCall = "https://{0}/api/sn_chg_rest/change/normal".format(self.snInstance)
payload = {
"category": category,
"u_subcategory": subCategory,
"u_change_manager": chngMngrID,
"assigned_to": userID,
"u_change_coordinator": chngCoordID,
"assignment_group": grpID,
"short_description": shortDesc,
"description": desc,
"justification": justDesc,
"implementation_plan": implmntPlan,
"risk_impact_analysis": riskImpact,
"backout_plan": backoutPlan,
"test_plan": testPlan,
"start_date": sTime,
"end_date": eTime,
"cab_required": False,
}
encoded_payload=json.dumps(payload)
changeResponse = requests.post(apiCall, auth=(self.user, self.pwd), headers=self.headers, data=encoded_payload)
changeData = changeResponse.json()
return changeData
def CreateStandardChange(self, templateName, grpName, assignee, coordinator, approver, category, subcategory, txtShortDesc, txtJustification):
templateID = self.getStandardChangeTemplateID(templateName)
grpID = self.getGroupID(grpName)
userID = self.getUserID(assignee)
chngCoordID = self.getUserID(coordinator)
chngMngrID = self.getUserID(approver)
sTime = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
eTime = (datetime.now()+timedelta(minutes=+1)).strftime("%Y-%m-%d %H:%M:%S")
apiCall = "https://{0}/api/sn_chg_rest/change/standard/{1}".format(self.snInstance,templateID)
payload={
"category": category,
"u_subcategory": subcategory,
"u_change_manager": chngMngrID,
"assigned_to": userID,
"u_change_coordinator": chngCoordID,
"assignment_group": grpID,
"short_description": txtShortDesc,
"justification": txtJustification,
"start_date": sTime,
"end_date": eTime
}
encoded_payload=json.dumps(payload)
apiResponse = requests.post(apiCall, auth=(self.user, self.pwd), headers=self.headers, data=encoded_payload)
return apiResponse.json()
######################################### END CLASS ################################