¡@
OpenStack Study: utils.py**** CubicPower OpenStack Study **** def set_db_item_state(context, neutron_item, new_state): with context.session.begin(subtransactions=True): if neutron_item["status"] != new_state: neutron_item["status"] = new_state context.session.merge(neutron_item)
**** CubicPower OpenStack Study **** def retrieve_subnet(context, subnet_id): return (context.session.query( models_v2.Subnet).filter(models_v2.Subnet.id == subnet_id).one())
**** CubicPower OpenStack Study **** def retrieve_ip_allocation_info(context, neutron_port): """Retrieves ip allocation info for a specific port if any.""" try: subnet_id = neutron_port["fixed_ips"][0]["subnet_id"] except (KeyError, IndexError): LOG.info(_("No ip allocation set")) return subnet = retrieve_subnet(context, subnet_id) allocated_ip = neutron_port["fixed_ips"][0]["ip_address"] is_gw_port = (neutron_port["device_owner"] == constants.DEVICE_OWNER_ROUTER_GW) gateway_ip = subnet["gateway_ip"] ip_allocation_info = h_info.IpAllocationInfo( is_gw=is_gw_port, ip_version=subnet["ip_version"], prefix=subnet["cidr"].split("/")[1], ip_address=allocated_ip, port_id=neutron_port["id"], gateway_ip=gateway_ip) return ip_allocation_info
**** CubicPower OpenStack Study **** def retrieve_nat_info(context, fip, fixed_prefix, floating_prefix, router): nat_info = h_info.NatInfo(source_address=fip["floating_ip_address"], source_prefix=floating_prefix, destination_address=fip["fixed_ip_address"], destination_prefix=fixed_prefix, floating_ip_id=fip["id"], fixed_port_id=fip["port_id"]) return nat_info
|