How to use nova - 10 common examples

def test_instance_flavor(self):
        inst = nova_notifier.Instance(context, self.instance)
        self.assertEqual(inst.flavor['name'], 'm1.tiny')
        self.assertEqual(inst.flavor['flavor_id'], '1')

msg = _("Requested network does not contain "
                            "enough (2+) usable hosts")
                    raise exc.HTTPBadRequest(explanation=msg)
            except netexc.AddrFormatError:
                msg = _("CIDR is malformed.")
                raise exc.HTTPBadRequest(explanation=msg)
            except netexc.AddrConversionError:
                msg = _("Address could not be converted.")
                raise exc.HTTPBadRequest(explanation=msg)

        networks = []
        try:
            if CONF.enable_network_quota:
                reservation = QUOTAS.reserve(context, networks=1)
        except exception.OverQuota:
            msg = _("Quota exceeded, too many networks.")
            raise exc.HTTPBadRequest(explanation=msg)

        kwargs['project_id'] = context.project_id

        try:
            networks = self.network_api.create(context,
                                               label=label, **kwargs)
            if CONF.enable_network_quota:
                QUOTAS.commit(context, reservation)
        except exception.PolicyNotAuthorized as e:
            raise exc.HTTPForbidden(explanation=six.text_type(e))
        except exception.CidrConflict as e:
            raise exc.HTTPConflict(explanation=e.format_message())
        except Exception:
            if CONF.enable_network_quota:
                QUOTAS.rollback(context, reservation)

def attach_volume(self, context,
                      volume_id,
                      instance_id,
                      device, **kwargs):
        validate_ec2_id(instance_id)
        validate_ec2_id(volume_id)
        volume_id = ec2utils.ec2_vol_id_to_uuid(volume_id)
        instance_uuid = ec2utils.ec2_inst_id_to_uuid(context, instance_id)
        instance = self.compute_api.get(context, instance_uuid,
                                        want_objects=True)
        LOG.audit(_('Attach volume %(volume_id)s to instance %(instance_id)s '
                    'at %(device)s'),
                  {'volume_id': volume_id,
                   'instance_id': instance_id,
                   'device': device},
                  context=context)

        self.compute_api.attach_volume(context, instance, volume_id, device)
        volume = self.volume_api.get(context, volume_id)
        ec2_attach_status = ec2utils.status_to_ec2_attach_status(volume)

        return {'attachTime': volume['attach_time'],
                'device': volume['mountpoint'],
                'instanceId': ec2utils.id_to_ec2_inst_id(instance_uuid),
                'requestId': context.request_id,
                'status': ec2_attach_status,
                'volumeId': ec2utils.id_to_ec2_vol_id(volume_id)}

def _mount_nfs(self, mount_path, nfs_share, options=None, ensure=False):
        """Mount nfs export to mount path."""
        utils.execute('mkdir', '-p', mount_path)

        # Construct the NFS mount command.
        nfs_cmd = ['mount', '-t', 'nfs']
        if CONF.libvirt.nfs_mount_options is not None:
            nfs_cmd.extend(['-o', CONF.libvirt.nfs_mount_options])
        if options:
            nfs_cmd.extend(options.split(' '))
        nfs_cmd.extend([nfs_share, mount_path])

        try:
            utils.execute(*nfs_cmd, run_as_root=True)
        except processutils.ProcessExecutionError as exc:
            if ensure and 'already mounted' in six.text_type(exc):
                LOG.warning(_LW("%s is already mounted"), nfs_share)
            else:
                raise

def _mount_glusterfs(self, mount_path, glusterfs_share,
                         options=None, ensure=False):
        """Mount glusterfs export to mount path."""
        utils.execute('mkdir', '-p', mount_path)

        gluster_cmd = ['mount', '-t', 'glusterfs']
        if options is not None:
            gluster_cmd.extend(options.split(' '))
        gluster_cmd.extend([glusterfs_share, mount_path])

        try:
            utils.execute(*gluster_cmd, run_as_root=True)
        except processutils.ProcessExecutionError as exc:
            if ensure and 'already mounted' in six.text_type(exc):
                LOG.warning(_LW("%s is already mounted"), glusterfs_share)
            else:
                raise

def _list_cobalt_hosts(self, context, availability_zone=None):
        """ Returns a list of all the hosts known to openstack running the cobalt service. """
        admin_context = context.elevated()
        services = self.db.service_get_all_by_topic(admin_context, CONF.cobalt_topic)

        if availability_zone is not None and ':' in availability_zone:
            parts = availability_zone.split(':')
            if len(parts) > 2:
                raise exception.NovaException(_('Invalid availability zone'))
            az = parts[0]
            host = parts[1]
            if (az, host) in [(srv['availability_zone'], srv['host']) for srv in services]:
                return [host]
            else:
                return []

        hosts = []
        for srv in services:
            in_availability_zone =  availability_zone is None or \
                                    availability_zone == \
                                            availability_zones.get_host_availability_zone(context,srv['host'])

            if srv['host'] not in hosts and in_availability_zone:
                hosts.append(srv['host'])
        return hosts

if not force_update:
                minimum = settable_quotas[key]['minimum']
                maximum = settable_quotas[key]['maximum']
                self._validate_quota_limit(key, value, minimum, maximum)
            valid_quotas[key] = value

        # NOTE(dims): Pass #2 - At this point we know that all the
        # values are correct and we can iterate and update them all in one
        # shot without having to worry about rolling back etc as we have done
        # the validation up front in the loop above.
        for key, value in valid_quotas.items():
            try:
                objects.Quotas.create_limit(context, project_id,
                                            key, value, user_id=user_id)
            except exception.QuotaExists:
                objects.Quotas.update_limit(context, project_id,
                                            key, value, user_id=user_id)
        # Note(gmann): Removed 'id' from update's response to make it same
        # as V2. If needed it can be added with microversion.
        return self._format_quota_set(
            None,
            self._get_quotas(context, id, user_id=user_id),
            filtered_quotas=filtered_quotas)

if not group_name and not group_id:
            err = "Not enough parameters, need group_name or group_id"
            raise exception.ApiError(_(err))
        notfound = exception.SecurityGroupNotFound
        if group_name:
            security_group = db.security_group_get_by_name(context,
                                                           context.project_id,
                                                           group_name)
            if not security_group:
                raise notfound(security_group_id=group_name)
        elif group_id:
            security_group = db.security_group_get(context, group_id)
            if not security_group:
                raise notfound(security_group_id=group_id)
        LOG.audit(_("Delete security group %s"), group_name, context=context)
        db.security_group_destroy(context, security_group.id)
        return True

class FloatingIpPoolNotFound(NotFound):
    msg_fmt = _("Floating IP pool not found.")
    safe = True


class NoMoreFloatingIps(FloatingIpNotFound):
    msg_fmt = _("Zero floating IPs available.")
    safe = True


class FloatingIpAssociated(NovaException):
    msg_fmt = _("Floating IP %(address)s is associated.")


class FloatingIpNotAssociated(NovaException):
    msg_fmt = _("Floating IP %(address)s is not associated.")


class NoFloatingIpsDefined(NotFound):
    msg_fmt = _("Zero floating IPs exist.")


class NoFloatingIpInterface(NotFound):
    msg_fmt = _("Interface %(interface)s not found.")


class FloatingIpAllocateFailed(NovaException):
    msg_fmt = _("Floating IP allocate failed.")


class FloatingIpAssociateFailed(NovaException):
    msg_fmt = _("Floating IP %(address)s association has failed.")

# be rolled up at the end of the check() function. These functions are
    # intended to be run in order and build on top of each other so order
    # matters.
    _upgrade_checks = (
        # Added in Ocata
        (_('Cells v2'), _check_cellsv2),
        # Added in Ocata
        (_('Placement API'), _check_placement),
        # Added in Rocky (but also useful going back to Pike)
        (_('Ironic Flavor Migration'), _check_ironic_flavor_migration),
        # Added in Rocky
        (_('Request Spec Migration'), _check_request_spec_migration),
        # Added in Stein (but also useful going back to Rocky)
        (_('Console Auths'), _check_console_auths),
        # Added in Train
        (_('Cinder API'), _check_cinder),
    )


CATEGORIES = {
    'upgrade': UpgradeCommands,
}


add_command_parsers = functools.partial(cmd_common.add_command_parsers,
                                        categories=CATEGORIES)


category_opt = cfg.SubCommandOpt('category',
                                 title='Command categories',
                                 help='Available categories',
                                 handler=add_command_parsers)