def put(self, context, id, request_data):
"""Update existing device values."""
net_obj = dbapi.network_devices_update(context, id, request_data)
device = jsonutils.to_primitive(net_obj)
utils.add_up_link(context, device)
return device, 200, None
python类to_primitive()的实例源码
def get(self, context, request_args, pagination_params):
"""Get all network interfaces."""
interfaces_obj, link_params = dbapi.network_interfaces_get_all(
context, request_args, pagination_params,
)
links = base.links_from(link_params)
response_body = {'network_interfaces': interfaces_obj, 'links': links}
return jsonutils.to_primitive(response_body), 200, None
def get(self, context, id):
"""Get network interface by given id"""
obj = dbapi.network_interfaces_get_by_id(context, id)
interface = jsonutils.to_primitive(obj)
interface['variables'] = jsonutils.to_primitive(obj.variables)
return interface, 200, None
def put(self, context, id, request_data):
"""Update existing network interface values."""
net_obj = dbapi.network_interfaces_update(context, id, request_data)
return jsonutils.to_primitive(net_obj), 200, None
def get(self, context, id):
"""Get a project details by id. Requires super admin privileges."""
project_obj = dbapi.projects_get_by_id(context, id)
project = jsonutils.to_primitive(project_obj)
project['variables'] = jsonutils.to_primitive(project_obj.variables)
return project, 200, None
def serialize_entity(context, entity):
return jsonutils.to_primitive(entity, convert_instances=True)
def serialize_entity(context, entity):
return jsonutils.to_primitive(entity, convert_instances=True)
def ping(self, context, arg, timeout=None):
arg_p = jsonutils.to_primitive(arg)
cctxt = self.client.prepare(timeout=timeout)
return cctxt.call(context, 'ping', arg=arg_p)
def ping(self, context, arg):
resp = {'service': self.service_name, 'arg': arg}
return jsonutils.to_primitive(resp)
def create_template(self, context, request_spec):
request_spec_p = jsonutils.to_primitive(request_spec)
return self.call(context, self.make_msg('create_template',
request_spec=request_spec_p))
def create_experiment(self, context, request_spec):
request_spec_p = jsonutils.to_primitive(request_spec)
return self.cast(context, self.make_msg('create_experiment',
request_spec=request_spec_p))
def create_dataset(self, context, request_spec):
request_spec_p = jsonutils.to_primitive(request_spec)
return self.cast(context, self.make_msg('create_dataset',
request_spec=request_spec_p))
def create_model(self, context, request_spec):
request_spec_p = jsonutils.to_primitive(request_spec)
return self.cast(context, self.make_msg('create_model',
request_spec=request_spec_p))
def load_model(self, context, request_spec):
request_spec_p = jsonutils.to_primitive(request_spec)
return self.cast(context, self.make_msg('load_model',
request_spec=request_spec_p))
def create_model_evaluation(self, context, request_spec):
request_spec_p = jsonutils.to_primitive(request_spec)
return self.cast(context, self.make_msg('create_model_evaluation',
request_spec=request_spec_p))
def create_learning(self, context, request_spec):
request_spec_p = jsonutils.to_primitive(request_spec)
return self.cast(context, self.make_msg('create_learning',
request_spec=request_spec_p))
def create_online_learning(self, context, request_spec):
request_spec_p = jsonutils.to_primitive(request_spec)
return self.call(context, self.make_msg('create_online_learning',
request_spec=request_spec_p))
def _notify(self, priority, ctxt, event_type, payload):
payload = self._serializer.serialize_entity(ctxt, payload)
# NOTE(sileht): simulate the kombu serializer
# this permit to raising an exception if something have not
# been serialized correctly
jsonutils.to_primitive(payload)
msg = dict(publisher_id=self.publisher_id,
priority=priority,
event_type=event_type,
payload=payload)
NOTIFICATIONS.append(msg)
def serialize_entity(context, entity):
return jsonutils.to_primitive(entity, convert_instances=True)
def serialize_entity(context, entity):
return json.to_primitive(entity, convert_instances=True)