Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def _store_record(record):
"""Update existing record via `control_number` or create new."""
if "control_number" in record:
existing_record = get_record(record['control_number'])
if existing_record is not None:
patch = jsonpatch.JsonPatch.from_diff(existing_record, record)
updated_record = existing_record.patch(patch=patch)
updated_record.commit()
else:
# New record with some hardcoded recid/control_number
create_record(data=record)
else:
create_record(data=record)
def _create_patch(self, spec, modified_spec, message='Patching pod'):
patch = jsonpatch.JsonPatch.from_diff(spec, modified_spec)
return jsonify(
{
"response": {
"allowed": True,
"status": {"message": message},
"patch": base64.b64encode(str(patch).encode()).decode(),
"patchtype": "JSONPatch",
}
server_config = dict(
driver=server['driver'],
properties=server['properties'],
driver_info=server['driver_info'],
name=server['name'],
)
# Add the pre-existing chassis_uuid only if
# it is present in the server configuration.
if hasattr(server, 'chassis_uuid'):
server_config['chassis_uuid'] = server['chassis_uuid']
# Note(TheJulia): If a password is defined and concealed, a
# patch will always be generated and re-asserted.
patch = jsonpatch.JsonPatch.from_diff(server_config, kwargs)
if not patch:
_exit_node_not_updated(module, server)
elif _choose_if_password_only(module, list(patch)):
# Note(TheJulia): Normally we would allow the general
# exception catch below, however this allows a specific
# message.
try:
server = cloud.patch_machine(
server['uuid'],
list(patch))
except Exception as e:
module.fail_json(msg="Failed to update node, "
"Error: %s" % e.message)
# Enumerate out a list of changed paths.
def update_to_patch(subject, update):
return [
item for item in jsonpatch.JsonPatch.from_diff(
subject,
update
) if update_patch_filter(update, item)
]
def diff(self, source, target):
source, target = (
self.sanitize_revision(source), self.sanitize_revision(target))
patch = jsonpatch.JsonPatch.from_diff(source, target)
return list(patch)
new_config['instance_uuid'] = instance_uuid
if properties:
machine_config['properties'] = machine['properties']
new_config['properties'] = properties
except KeyError as e:
self.log.debug(
"Unexpected machine response missing key %s [%s]",
e.args[0], name_or_id)
raise OpenStackCloudException(
"Machine update failed - machine [%s] missing key %s. "
"Potential API issue."
% (name_or_id, e.args[0]))
try:
patch = jsonpatch.JsonPatch.from_diff(machine_config, new_config)
except Exception as e:
raise OpenStackCloudException(
"Machine update failed - Error generating JSON patch object "
"for submission to the API. Machine: %s Error: %s"
% (name_or_id, str(e)))
with _utils.shade_exceptions(
"Machine update failed - patch operation failed on Machine "
"{node}".format(node=name_or_id)
):
if not patch:
return dict(
node=machine,
changes=None
)
else: