Python oslo_serialization.jsonutils 模块，to_primitive() 实例源码

def get(self, context, request_args, pagination_params):
        """Get all users. Requires project admin privileges."""
        user_id = request_args["id"]
        user_name = request_args["name"]

        if user_id:
            user_obj = dbapi.users_get_by_id(context, user_id)
            user_obj.data = user_obj.variables
            users_obj = [user_obj]
            link_params = {}

        if user_name:
            users_obj, link_params = dbapi.users_get_by_name(
                context, user_name, request_args, pagination_params,
            )
        else:
            users_obj, link_params = dbapi.users_get_all(
                context, request_args, pagination_params,
            )
        links = base.links_from(link_params)
        response_body = {'users': users_obj, 'links': links}
        return jsonutils.to_primitive(response_body), 200, None

def post(self, context, request_data):
        """Create a new user. Requires project admin privileges."""
        # NOTE(sulo): Instead of using context project_id from
        # header, here we always ensure, user create gets project_id
        # from request param.
        project_id = request_data["project_id"]
        dbapi.projects_get_by_id(context, project_id)
        api_key = uuidutils.generate_uuid()
        request_data["api_key"] = api_key
        user_obj = dbapi.users_create(context, request_data)

        location = v1.api.url_for(
            UserById, id=user_obj.id, _external=True
        )
        headers = {'Location': location}

        return jsonutils.to_primitive(user_obj), 201, headers

def post(self, context, request_data):
        """Create a new region."""
        json = util.copy_project_id_into_json(context, request_data)
        region_obj = dbapi.regions_create(context, json)
        region = jsonutils.to_primitive(region_obj)
        if 'variables' in json:
            region["variables"] = jsonutils.to_primitive(region_obj.variables)
        else:
            region["variables"] = {}

        location = v1.api.url_for(
            RegionsById, id=region_obj.id, _external=True
        )
        headers = {'Location': location}

        return region, 201, headers

def get(self, context, request_args):
        region_id = request_args["region_id"]
        cell_id = request_args["cell_id"]

        filters = {}
        if region_id:
            filters['region_id'] = region_id

        # TODO(sulo): allow other filters based on services
        if cell_id:
            filters['cell_id'] = cell_id

        try:
            hosts_obj = dbapi.hosts_get_all(context, filters)
        except exceptions.NotFound:
            return self.error_response(404, 'Not Found')
        except Exception as err:
            LOG.error("Error during host get: %s" % err)
            return self.error_response(500, 'Unknown Error')

        _inventory = self.generate_ansible_inventory(hosts_obj)
        inventory = jsonutils.to_primitive(_inventory)
        return inventory, 200, None

def post(self, context, request_data):
        """Create a new cloud."""
        json = util.copy_project_id_into_json(context, request_data)
        cloud_obj = dbapi.clouds_create(context, json)
        cloud = jsonutils.to_primitive(cloud_obj)
        if 'variables' in json:
            cloud["variables"] = jsonutils.to_primitive(cloud_obj.variables)
        else:
            cloud["variables"] = {}

        location = v1.api.url_for(
            CloudsById, id=cloud_obj.id, _external=True
        )
        headers = {'Location': location}

        return cloud, 201, headers

def post(self, context, request_data):
        """Create a new host."""
        json = util.copy_project_id_into_json(context, request_data)
        host_obj = dbapi.hosts_create(context, json)
        host = jsonutils.to_primitive(host_obj)
        if 'variables' in json:
            host["variables"] = jsonutils.to_primitive(host_obj.variables)
        else:
            host["variables"] = {}

        utils.add_up_link(context, host)

        location = v1.api.url_for(
            HostById, id=host_obj.id, _external=True
        )
        headers = {'Location': location}

        return host, 201, headers

def post(self, context, request_data):
        """Create a new network."""
        json = util.copy_project_id_into_json(context, request_data)
        network_obj = dbapi.networks_create(context, json)
        network = jsonutils.to_primitive(network_obj)
        if 'variables' in json:
            network["variables"] = jsonutils.to_primitive(
                network_obj.variables)
        else:
            network["variables"] = {}

        location = v1.api.url_for(
            NetworkById, id=network_obj.id, _external=True
        )
        headers = {'Location': location}

        return network, 201, headers

def get(self, context, request_args, pagination_params):
        """Get all network devices."""
        details = request_args.get("details")
        devices_obj, link_params = dbapi.network_devices_get_all(
            context, request_args, pagination_params,
        )
        if details:
            devices_obj = [utils.get_resource_with_vars(request_args, d)
                           for d in devices_obj]

        links = base.links_from(link_params)
        response_body = jsonutils.to_primitive(
            {'network_devices': devices_obj, 'links': links}
        )

        for device in response_body["network_devices"]:
            utils.add_up_link(context, device)

        return response_body, 200, None

def post(self, context, request_data):
        """Create a new network device."""
        json = util.copy_project_id_into_json(context, request_data)
        obj = dbapi.network_devices_create(context, json)
        device = jsonutils.to_primitive(obj)
        if 'variables' in json:
            device["variables"] = jsonutils.to_primitive(obj.variables)
        else:
            device["variables"] = {}

        utils.add_up_link(context, device)

        location = v1.api.url_for(
            NetworkDeviceById, id=obj.id, _external=True
        )
        headers = {'Location': location}

        return device, 201, headers

def get(self, context, request_args, pagination_params):
        """Get all projects. Requires super admin privileges."""
        project_name = request_args["name"]
        details = request_args.get("details")

        if project_name:
            projects_obj, link_params = dbapi.projects_get_by_name(
                context, project_name, request_args, pagination_params,
            )
        else:
            projects_obj, link_params = dbapi.projects_get_all(
                context, request_args, pagination_params,
            )
            if details:
                projects_obj = [utils.get_resource_with_vars(request_args, p)
                                for p in projects_obj]

        links = base.links_from(link_params)
        response_body = {'projects': projects_obj, 'links': links}
        return jsonutils.to_primitive(response_body), 200, None

def post(self, context, request_data):
        """Create a new project. Requires super admin privileges."""
        project_obj = dbapi.projects_create(context, request_data)

        location = v1.api.url_for(
            ProjectById, id=project_obj.id, _external=True
        )
        headers = {'Location': location}

        project = jsonutils.to_primitive(project_obj)
        if 'variables' in request_data:
            project["variables"] = \
                jsonutils.to_primitive(project_obj.variables)
        else:
            project["variables"] = {}
        return project, 201, headers

def serialize(self, entity):
        if entity is None:
            return None

        key = self._get_serialization_key(type(entity))

        # Primitive or not registered type.
        if not key:
            return jsonutils.dumps(
                jsonutils.to_primitive(entity, convert_instances=True)
            )

        serializer = self.serializers.get(key)

        if not serializer:
            raise RuntimeError(
                "Failed to find a serializer for the key: %s" % key
            )

        result = {
            '__serial_key': key,
            '__serial_data': serializer.serialize(entity)
        }

        return jsonutils.dumps(result)

def select_destinations(self, ctxt,
                            request_spec=None, filter_properties=None,
                            spec_obj=_sentinel):
        """Returns destinations(s) best suited for this RequestSpec.

        The result should be a list of dicts with 'host', 'nodename' and
        'limits' as keys.
        """

        # TODO(sbauza): Change the method signature to only accept a spec_obj
        # argument once API v5 is provided.
        if spec_obj is self._sentinel:
            spec_obj = objects.RequestSpec.from_primitives(ctxt,
                                                           request_spec,
                                                           filter_properties)
        dests = self.driver.select_destinations(ctxt, spec_obj)
        return jsonutils.to_primitive(dests)

def allocate_for_instance(self, ctxt, instance_id, project_id, host,
                              rxtx_factor, vpn, requested_networks, macs=None,
                              dhcp_options=None):
        version = '1.13'
        if not self.client.can_send_version(version):
            version = '1.9'
            if requested_networks:
                requested_networks = requested_networks.as_tuples()

        if CONF.multi_host:
            cctxt = self.client.prepare(version=version, server=host)
        else:
            cctxt = self.client.prepare(version=version)
        return cctxt.call(ctxt, 'allocate_for_instance',
                          instance_id=instance_id, project_id=project_id,
                          host=host, rxtx_factor=rxtx_factor, vpn=vpn,
                          requested_networks=requested_networks,
                          macs=jsonutils.to_primitive(macs))

def test_from_request_spec_instance_props(self):
        requests = objects.InstancePCIRequests(
            requests=[objects.InstancePCIRequest(count=1,
                                                 request_id=FAKE_UUID,
                                                 spec=[{'vendor_id': '8086',
                                                        'device_id': '1502'}])
                      ],
            instance_uuid=FAKE_UUID)
        result = jsonutils.to_primitive(requests)
        result = objects.InstancePCIRequests.from_request_spec_instance_props(
                                                                        result)
        self.assertEqual(1, len(result.requests))
        self.assertEqual(1, result.requests[0].count)
        self.assertEqual(FAKE_UUID, result.requests[0].request_id)
        self.assertEqual([{'vendor_id': '8086', 'device_id': '1502'}],
                          result.requests[0].spec)

def setUp(self):
        super(ComputeRpcAPITestCase, self).setUp()
        self.context = context.get_admin_context()
        self.fake_flavor_obj = fake_flavor.fake_flavor_obj(self.context)
        self.fake_flavor = jsonutils.to_primitive(self.fake_flavor_obj)
        instance_attr = {'host': 'fake_host',
                         'instance_type_id': self.fake_flavor_obj['id'],
                         'instance_type': self.fake_flavor_obj}
        self.fake_instance_obj = fake_instance.fake_instance_obj(self.context,
                                                   **instance_attr)
        self.fake_instance = jsonutils.to_primitive(self.fake_instance_obj)
        self.fake_volume_bdm = objects_block_dev.BlockDeviceMapping(
                **fake_block_device.FakeDbBlockDeviceDict(
                    {'source_type': 'volume', 'destination_type': 'volume',
                     'instance_uuid': self.fake_instance_obj.uuid,
                     'volume_id': 'fake-volume-id'}))

def get(self, context, id):
        """Get a user details by id. Requires project admin privileges."""
        user_obj = dbapi.users_get_by_id(context, id)
        return jsonutils.to_primitive(user_obj), 200, None

def get(self, context, resources, id, request_args=None):
        """Get variables for given resource."""
        obj = dbapi.resource_get_by_id(context, resources, id)
        obj = utils.format_variables(request_args, obj)
        resp = {"variables": jsonutils.to_primitive(obj.vars)}
        return resp, 200, None

def put(self, context, resources, id, request_data):
        """
        Update existing resource variables, or create if it does
        not exist.
        """
        obj = dbapi.variables_update_by_resource_id(
            context, resources, id, request_data
        )
        resp = {"variables": jsonutils.to_primitive(obj.variables)}
        return resp, 200, None

def put(self, context, id, request_data):
        """Update existing region."""
        region_obj = dbapi.regions_update(context, id, request_data)
        return jsonutils.to_primitive(region_obj), 200, None

def get(self, context, request_args, pagination_params):
        """Get all devices, with optional filtering."""
        details = request_args.get("details")
        device_objs, link_params = dbapi.devices_get_all(
            context, request_args, pagination_params,
        )
        links = base.links_from(link_params)

        devices = {"hosts": [], "network-devices": []}
        for device_obj in device_objs:
            if details:
                device = utils.get_resource_with_vars(request_args,
                                                      device_obj)
            else:
                device = jsonutils.to_primitive(device_obj)

            utils.add_up_link(context, device)

            if isinstance(device_obj, models.Host):
                devices["hosts"].append(device)
            elif isinstance(device_obj, models.NetworkDevice):
                devices["network-devices"].append(device)
            else:
                LOG.error(
                    "The device is of unknown type: '%s'", device_obj
                )
                raise exceptions.UnknownException

        response_body = jsonutils.to_primitive(
            {'devices': devices, 'links': links}
        )

        return response_body, 200, None

def get(self, context, request_args, pagination_params):
        """Get all cells, with optional filtering."""
        details = request_args.get("details")

        cells_obj, link_params = dbapi.cells_get_all(
            context, request_args, pagination_params,
        )
        if details:
            cells_obj = [utils.get_resource_with_vars(request_args, cell)
                         for cell in cells_obj]

        links = base.links_from(link_params)
        response_body = {'cells': cells_obj, 'links': links}
        return jsonutils.to_primitive(response_body), 200, None

def put(self, context, id, request_data):
        """Update existing cell."""
        cell_obj = dbapi.cells_update(context, id, request_data)
        return jsonutils.to_primitive(cell_obj), 200, None

def get(self, context, request_args, pagination_params):
        """Get cloud(s) for the project. Get cloud details if
        for a particular cloud.
        """
        cloud_id = request_args.get("id")
        cloud_name = request_args.get("name")
        details = request_args.get("details")

        if not (cloud_id or cloud_name):
            # Get all clouds for this project
            clouds_obj, link_params = dbapi.clouds_get_all(
                context, request_args, pagination_params,
            )
            if details:
                clouds_obj = [utils.get_resource_with_vars(request_args, c)
                              for c in clouds_obj]
        else:
            if cloud_name:
                cloud_obj = dbapi.clouds_get_by_name(context, cloud_name)
                cloud_obj.data = cloud_obj.variables

            if cloud_id:
                cloud_obj = dbapi.clouds_get_by_id(context, cloud_id)
                cloud_obj.data = cloud_obj.variables

            clouds_obj = [cloud_obj]
            link_params = {}
        links = base.links_from(link_params)
        response_body = {'clouds': clouds_obj, 'links': links}
        return jsonutils.to_primitive(response_body), 200, None

def get(self, context, id):
        cloud_obj = dbapi.clouds_get_by_id(context, id)
        cloud = jsonutils.to_primitive(cloud_obj)
        cloud['variables'] = jsonutils.to_primitive(cloud_obj.variables)
        return cloud, 200, None

def put(self, context, id, request_data):
        """Update existing cloud."""
        cloud_obj = dbapi.clouds_update(context, id, request_data)
        return jsonutils.to_primitive(cloud_obj), 200, None

def put(self, context, id, request_data):
        """Update existing host data, or create if it does not exist."""
        host_obj = dbapi.hosts_update(context, id, request_data)

        host = jsonutils.to_primitive(host_obj)

        utils.add_up_link(context, host)

        return host, 200, None

def get(self, context, request_args, pagination_params):
        """Get all networks, with optional filtering."""
        details = request_args.get("details")
        networks_obj, link_params = dbapi.networks_get_all(
            context, request_args, pagination_params,
        )
        if details:
            networks_obj = [utils.get_resource_with_vars(request_args, n)
                            for n in networks_obj]

        links = base.links_from(link_params)
        response_body = {'networks': networks_obj, 'links': links}
        return jsonutils.to_primitive(response_body), 200, None

def get(self, context, id):
        """Get network by given id"""
        obj = dbapi.networks_get_by_id(context, id)
        device = jsonutils.to_primitive(obj)
        device['variables'] = jsonutils.to_primitive(obj.variables)
        return device, 200, None

def get(self, context, id, request_args):
        """Get network device by given id"""
        obj = dbapi.network_devices_get_by_id(context, id)
        obj = utils.format_variables(request_args, obj)
        device = jsonutils.to_primitive(obj)
        device['variables'] = jsonutils.to_primitive(obj.vars)

        utils.add_up_link(context, device)

        return device, 200, None

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

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)