Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
raise RallyRESTAPIError('Target %s %s could not be located' % (entityName, formattedID))
target = response.next()
oid = target.ObjectID
itemData['ObjectID'] = oid
resource = '%s/%s?key=%s' % (entityName.lower(), oid, auth_token)
context, augments = self.contextHelper.identifyContext(workspace=workspace, project=project)
if augments:
resource += ("&" + "&".join(augments))
full_resource_url = "%s/%s" % (self.service_url, resource)
itemData = self.validateAttributeNames(entityName, itemData)
item = {entityName: self._greased(itemData)}
payload = json.dumps(item)
if self._log:
self._logDest.write('%s POST %s\n%27.27s %s\n' % (timestamp(), resource, " ", item))
self._logDest.flush()
response = self.session.post(full_resource_url, data=payload, headers=RALLY_REST_HEADERS)
response = RallyRESTResponse(self.session, context, resource, response, "shell", 0)
if response.status_code != HTTP_REQUEST_SUCCESS_CODE:
raise RallyRESTAPIError('Unable to update the %s' % entityName)
# now issue a request to get the entity item (mostly so we can get the FormattedID)
# and return it
item = self._itemQuery(entityName, oid, workspace=workspace, project=project)
return item
All optional keyword args:
fetch=True/False or "List,Of,Attributes,We,Are,Interested,In"
query='FieldName = "some value"' or ['fld1 = 19', 'fld27 != "Shamu"', etc.]
order="fieldName ASC|DESC"
instance=False/True
pagesize=n
start=n
limit=n
projectScopeUp=True/False
projectScopeDown=True/False
"""
context, resource, full_resource_url, limit = self._buildRequest(entity, fetch, query, order, kwargs)
if self._log:
# unquote the resource for enhanced readability
self._logDest.write('%s GET %s\n' % (timestamp(), unquote(resource)))
self._logDest.flush()
threads = 0
if 'threads' in kwargs:
if kwargs['threads'] in [1,2,3,4,5,6,7,8]:
threads = kwargs['threads']
else:
threads = 2
response = self._getRequestResponse(context, full_resource_url, limit, threads=threads)
if kwargs and 'instance' in kwargs and kwargs['instance'] == True and response.resultCount == 1:
return response.next()
return response
try:
# a response has status_code, content and data attributes
# the data attribute is a dict that has a single entry for the key 'QueryResult'
# or 'OperationResult' whose value is in turn a dict with values of
# 'Errors', 'Warnings', 'Results'
response = self.session.get(request_url, timeout=SERVICE_REQUEST_TIMEOUT)
except Exception as ex:
if response:
##
## print("Exception detected for session.get requests, response status code: %s" % response.status_code)
##
ret_code, content = response.status_code, response.content
else:
ret_code, content = PAGE_NOT_FOUND_CODE, str(ex.args[0])
if self._log:
self._logDest.write('%s %s\n' % (timestamp(), ret_code))
self._logDest.flush()
errorResponse = ErrorResponse(ret_code, content)
response = RallyRESTResponse(self.session, context, request_url, errorResponse, self.hydration, 0)
return response
##
## print("response.status_code is %s" % response.status_code)
##
if response.status_code != HTTP_REQUEST_SUCCESS_CODE:
if self._log:
code, verbiage = response.status_code, response.content[:56]
self._logDest.write('%s %s %s ...\n' % (timestamp(), code, verbiage))
self._logDest.flush()
##
## print(response)
problem = "ERRORS: %s\nWARNINGS: %s\n" % ("\n".join(response.errors),
"\n".join(response.warnings))
raise RallyRESTAPIError(problem)
##
## print(response.content)
##
response = RallyRESTResponse(self.session, context, resource, response, "shell", 0)
if response.errors:
status = False
desc = response.errors[0]
else:
status = True
desc = '%s deleted' % entityName
if self._log:
self._logDest.write('%s %s %s\n' % (timestamp(), response.status_code, desc))
self._logDest.flush()
return status
def _itemQuery(self, entityName, oid, workspace=None, project=None):
"""
Internal method to retrieve a specific instance of an entity identified by the OID.
"""
##
## print("Rally._itemQuery('%s', %s, workspace=%s, project=%s)" % (entityName, oid, workspace, project))
##
resource = '%s/%s' % (entityName, oid)
context, augments = self.contextHelper.identifyContext(workspace=workspace, project=project)
if augments:
resource += ("?" + "&".join(augments))
if self._log:
self._logDest.write('%s GET %s\n' % (timestamp(), resource))
self._logDest.flush()
response = self._getResourceByOID(context, entityName, oid)
if self._log:
self._logDest.write('%s %s %s\n' % (timestamp(), response.status_code, resource))
self._logDest.flush()
if not response or response.status_code != HTTP_REQUEST_SUCCESS_CODE:
problem = "Unreferenceable %s OID: %s" % (entityName, oid)
raise RallyRESTAPIError('%s %s' % (response.status_code, problem))
response = RallyRESTResponse(self.session, context, '%s.x' % entityName, response, "full", 1)
item = response.next()
return item # return back an instance representing the item
if '_disableAugments' not in kwargs:
contextDict = context.asDict()
##
## print("_getResourceByOID, current contextDict: %s" % repr(contextDict))
## sys.stdout.flush()
##
context, augments = self.contextHelper.identifyContext(**contextDict)
if augments:
resource += ("?" + "&".join(augments))
##
## print("_getResourceByOID, modified contextDict: %s" % repr(context.asDict()))
## sys.stdout.flush()
##
full_resource_url = "%s/%s" % (self.service_url, resource)
if self._logAttrGet:
self._logDest.write('%s GET %s\n' % (timestamp(), resource))
self._logDest.flush()
##
## print("issuing GET for resource: %s" % full_resource_url)
## sys.stdout.flush()
##
try:
raw_response = self.session.get(full_resource_url)
except Exception as ex:
exctype, value, tb = sys.exc_info()
warning("%s: %s" % (exctype, value))
return None
##
## print("_getResourceByOID(%s, %s) raw_response: %s" % (entity, oid, raw_response))
## sys.stdout.flush()
##
return raw_response
elif type(dest) == bytes:
try:
mode = 'w'
if append:
mode = 'a'
self._logDest = open(dest, mode)
except IOError as ex:
self._log = False
self._logDest = None
else:
self._log = False
# emit a warning that logging is disabled due to a faulty dest arg
warning('Logging dest arg cannot be written to, proceeding with logging disabled.')
if self._log:
scopeNote = '%s Following entries record Rally REST API interaction via %s for user: %s' % \
(timestamp(), self.service_url, self.user)
self._logDest.write('%s\n' % scopeNote)
self._logDest.flush()
if attrget:
self._logAttrGet = True