Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def update(self, dataset_id, dataset_update):
url = '{base}/{dataset_id}'.format(
base=URL_BASE, dataset_id=dataset_id)
return self._update(url, HTTPMethod.PUT, requests.codes.ok,
dataset_update, DATA_SET_DESC)
def add_user(self, group_id, user_id):
url = self._base(group_id) + '/users/' + str(user_id)
desc = "a User in a Group"
return self._update(url, HTTPMethod.PUT, requests.codes.no_content, {}, desc)
def _update(self, url, method, success_code, obj_update, obj_desc):
if method == HTTPMethod.PUT:
response = self.transport.put(url, obj_update)
elif method == HTTPMethod.PATCH:
response = self.transport.patch(url, obj_update)
if response.status_code == success_code:
if str(response.text) == '':
return
else:
obj = response.json()
self.print_json("Updated", obj)
return obj
else:
self.logger.debug("Error updating " + obj_desc + ": "
+ self.transport.dump_response(response))
raise Exception("Error updating " + obj_desc + ": "
+ response.text)
def update(self, group_id, group_update):
return self._update(self._base(group_id), HTTPMethod.PUT, requests.codes.ok, group_update, self.groupDesc)
def put(self, url, body):
headers = self._headers_send_json()
return self.request(url, HTTPMethod.PUT, headers, {},
self._obj_to_json(body))
def commit_execution(self, stream_id, execution_id):
url = self._base(stream_id) + '/executions/' + str(execution_id) + '/commit'
return self._update(url, HTTPMethod.PUT, requests.codes.ok, {}, self.executionDesc)
def update_pdp(self, dataset_id, policy_id, policy_update):
url = '{base}/{dataset_id}/policies/{policy_id}'.format(
base=URL_BASE, dataset_id=dataset_id, policy_id=policy_id)
return self._update(url, HTTPMethod.PUT, requests.codes.ok,
policy_update, PDP_DESC)