@utils.exc_wrapper(internal=True)<EOL>def create_port(context, port):
LOG.info("<STR_LIT>" % context.tenant_id)<EOL>port_attrs = port["<STR_LIT:port>"]<EOL>admin_only = ["<STR_LIT>", "<STR_LIT>", "<STR_LIT>", "<STR_LIT>",<EOL>"<STR_LIT>", "<STR_LIT>",<EOL>"<STR_LIT>"]<EOL>utils.filter_body(context, port_attrs, admin_only=admin_only)<EOL>port_attrs = port["<STR_LIT:port>"]<EOL>mac_address = utils.pop_param(port_attrs, "<STR_LIT>", None)<EOL>use_forbidden_mac_range = utils.pop_param(port_attrs,<EOL>"<STR_LIT>", False)<EOL>segment_id = utils.pop_param(port_attrs, "<STR_LIT>")<EOL>fixed_ips = utils.pop_param(port_attrs, "<STR_LIT>")<EOL>if "<STR_LIT>" not in port_attrs:<EOL><INDENT>port_attrs['<STR_LIT>'] = "<STR_LIT>"<EOL><DEDENT>device_id = port_attrs['<STR_LIT>']<EOL>if "<STR_LIT>" not in port_attrs:<EOL><INDENT>port_attrs['<STR_LIT>'] = "<STR_LIT>"<EOL><DEDENT>instance_node_id = port_attrs['<STR_LIT>']<EOL>net_id = port_attrs["<STR_LIT>"]<EOL>port_id = uuidutils.generate_uuid()<EOL>net = db_api.network_find(context=context, limit=None, sorts=['<STR_LIT:id>'],<EOL>marker=None, page_reverse=False, fields=None,<EOL>id=net_id, scope=db_api.ONE)<EOL>if not net:<EOL><INDENT>raise n_exc.NetworkNotFound(net_id=net_id)<EOL><DEDENT>_raise_if_unauthorized(context, net)<EOL>if device_id:<EOL><INDENT>existing_ports = db_api.port_find(context,<EOL>network_id=net_id,<EOL>device_id=device_id,<EOL>scope=db_api.ONE)<EOL>if existing_ports:<EOL><INDENT>raise n_exc.BadRequest(<EOL>resource="<STR_LIT:port>", msg="<STR_LIT>"<EOL>"<STR_LIT>")<EOL><DEDENT><DEDENT>if fixed_ips:<EOL><INDENT>quota.QUOTAS.limit_check(context, context.tenant_id,<EOL>fixed_ips_per_port=len(fixed_ips))<EOL><DEDENT>if not STRATEGY.is_provider_network(net_id):<EOL><INDENT>segment_id = None<EOL>port_count = db_api.port_count_all(context, network_id=[net_id],<EOL>tenant_id=[context.tenant_id])<EOL>quota.QUOTAS.limit_check(<EOL>context, context.tenant_id,<EOL>ports_per_network=port_count + <NUM_LIT:1>)<EOL><DEDENT>else:<EOL><INDENT>if not segment_id:<EOL><INDENT>raise q_exc.AmbiguousNetworkId(net_id=net_id)<EOL><DEDENT><DEDENT>network_plugin = utils.pop_param(port_attrs, "<STR_LIT>")<EOL>if not network_plugin:<EOL><INDENT>network_plugin = net["<STR_LIT>"]<EOL><DEDENT>port_attrs["<STR_LIT>"] = network_plugin<EOL>ipam_driver = _get_ipam_driver(net, port=port_attrs)<EOL>net_driver = _get_net_driver(net, port=port_attrs)<EOL>base_net_driver = _get_net_driver(net)<EOL>security_groups = utils.pop_param(port_attrs, "<STR_LIT>")<EOL>if security_groups is not None:<EOL><INDENT>raise q_exc.SecurityGroupsNotImplemented()<EOL><DEDENT>group_ids, security_groups = _make_security_group_list(context,<EOL>security_groups)<EOL>quota.QUOTAS.limit_check(context, context.tenant_id,<EOL>security_groups_per_port=len(group_ids))<EOL>addresses = []<EOL>backend_port = None<EOL>with utils.CommandManager().execute() as cmd_mgr:<EOL><INDENT>@cmd_mgr.do<EOL>def _allocate_ips(fixed_ips, net, port_id, segment_id, mac,<EOL>**kwargs):<EOL><INDENT>if fixed_ips:<EOL><INDENT>if (STRATEGY.is_provider_network(net_id) and<EOL>not context.is_admin):<EOL><INDENT>raise n_exc.NotAuthorized()<EOL><DEDENT>ips, subnets = split_and_validate_requested_subnets(context,<EOL>net_id,<EOL>segment_id,<EOL>fixed_ips)<EOL>kwargs["<STR_LIT>"] = ips<EOL>kwargs["<STR_LIT>"] = subnets<EOL><DEDENT>ipam_driver.allocate_ip_address(<EOL>context, addresses, net["<STR_LIT:id>"], port_id,<EOL>CONF.QUARK.ipam_reuse_after, segment_id=segment_id,<EOL>mac_address=mac, **kwargs)<EOL><DEDENT>@cmd_mgr.undo<EOL>def _allocate_ips_undo(addr, **kwargs):<EOL><INDENT>LOG.info("<STR_LIT>")<EOL>if addresses:<EOL><INDENT>for address in addresses:<EOL><INDENT>try:<EOL><INDENT>with context.session.begin():<EOL><INDENT>ipam_driver.deallocate_ip_address(context, address,<EOL>**kwargs)<EOL><DEDENT><DEDENT>except Exception:<EOL><INDENT>LOG.exception("<STR_LIT>" % address)<EOL><DEDENT><DEDENT><DEDENT><DEDENT>@cmd_mgr.do<EOL>def _allocate_mac(net, port_id, mac_address,<EOL>use_forbidden_mac_range=False,<EOL>**kwargs):<EOL><INDENT>mac = ipam_driver.allocate_mac_address(<EOL>context, net["<STR_LIT:id>"], port_id, CONF.QUARK.ipam_reuse_after,<EOL>mac_address=mac_address,<EOL>use_forbidden_mac_range=use_forbidden_mac_range, **kwargs)<EOL>return mac<EOL><DEDENT>@cmd_mgr.undo<EOL>def _allocate_mac_undo(mac, **kwargs):<EOL><INDENT>LOG.info("<STR_LIT>")<EOL>if mac:<EOL><INDENT>try:<EOL><INDENT>with context.session.begin():<EOL><INDENT>ipam_driver.deallocate_mac_address(context,<EOL>mac["<STR_LIT:address>"])<EOL><DEDENT><DEDENT>except Exception:<EOL><INDENT>LOG.exception("<STR_LIT>" % mac)<EOL><DEDENT><DEDENT><DEDENT>@cmd_mgr.do<EOL>def _allocate_backend_port(mac, addresses, net, port_id, **kwargs):<EOL><INDENT>backend_port = net_driver.create_port(<EOL>context, net["<STR_LIT:id>"],<EOL>port_id=port_id,<EOL>security_groups=group_ids,<EOL>device_id=device_id,<EOL>instance_node_id=instance_node_id,<EOL>mac_address=mac,<EOL>addresses=addresses,<EOL>base_net_driver=base_net_driver)<EOL>_filter_backend_port(backend_port)<EOL>return backend_port<EOL><DEDENT>@cmd_mgr.undo<EOL>def _allocate_back_port_undo(backend_port,<EOL>**kwargs):<EOL><INDENT>LOG.info("<STR_LIT>")<EOL>try:<EOL><INDENT>backend_port_uuid = None<EOL>if backend_port:<EOL><INDENT>backend_port_uuid = backend_port.get("<STR_LIT>")<EOL><DEDENT>net_driver.delete_port(context, backend_port_uuid)<EOL><DEDENT>except Exception:<EOL><INDENT>LOG.exception(<EOL>"<STR_LIT>" % backend_port)<EOL><DEDENT><DEDENT>@cmd_mgr.do<EOL>def _allocate_db_port(port_attrs, backend_port, addresses, mac,<EOL>**kwargs):<EOL><INDENT>port_attrs["<STR_LIT>"] = net["<STR_LIT:id>"]<EOL>port_attrs["<STR_LIT:id>"] = port_id<EOL>port_attrs["<STR_LIT>"] = security_groups<EOL>LOG.info("<STR_LIT>" % backend_port)<EOL>port_attrs.update(backend_port)<EOL>with context.session.begin():<EOL><INDENT>new_port = db_api.port_create(<EOL>context, addresses=addresses, mac_address=mac["<STR_LIT:address>"],<EOL>backend_key=backend_port["<STR_LIT>"], **port_attrs)<EOL><DEDENT>return new_port<EOL><DEDENT>@cmd_mgr.undo<EOL>def _allocate_db_port_undo(new_port,<EOL>**kwargs):<EOL><INDENT>LOG.info("<STR_LIT>")<EOL>if not new_port:<EOL><INDENT>return<EOL><DEDENT>try:<EOL><INDENT>with context.session.begin():<EOL><INDENT>db_api.port_delete(context, new_port)<EOL><DEDENT><DEDENT>except Exception:<EOL><INDENT>LOG.exception(<EOL>"<STR_LIT>" % backend_port)<EOL><DEDENT><DEDENT>mac = _allocate_mac(net, port_id, mac_address,<EOL>use_forbidden_mac_range=use_forbidden_mac_range)<EOL>_allocate_ips(fixed_ips, net, port_id, segment_id, mac)<EOL>backend_port = _allocate_backend_port(mac, addresses, net, port_id)<EOL>new_port = _allocate_db_port(port_attrs, backend_port, addresses, mac)<EOL><DEDENT>return v._make_port_dict(new_port)<EOL>
Create a port

    Create a port which is a connection point of a device (e.g., a VM
    NIC) to attach to a L2 Neutron network.
    : param context: neutron api request context
    : param port: dictionary describing the port, with keys
        as listed in the RESOURCE_ATTRIBUTE_MAP object in
        neutron/api/v2/attributes.py.  All keys will be populated.