¡@
OpenStack Study: db.py**** CubicPower OpenStack Study **** def get_network_bindings(session, network_id): session = session or db.get_session() return (session.query(models.TzNetworkBinding). filter_by(network_id=network_id). all())
**** CubicPower OpenStack Study **** def get_network_bindings_by_vlanid(session, vlan_id): session = session or db.get_session() return (session.query(models.TzNetworkBinding). filter_by(vlan_id=vlan_id). all())
**** CubicPower OpenStack Study **** def add_network_binding(session, network_id, binding_type, phy_uuid, vlan_id): with session.begin(subtransactions=True): binding = models.TzNetworkBinding(network_id, binding_type, phy_uuid, vlan_id) session.add(binding) return binding
**** CubicPower OpenStack Study **** def add_neutron_nsx_network_mapping(session, neutron_id, nsx_switch_id): with session.begin(subtransactions=True): mapping = models.NeutronNsxNetworkMapping( neutron_id=neutron_id, nsx_id=nsx_switch_id) session.add(mapping) return mapping
**** CubicPower OpenStack Study **** def add_neutron_nsx_port_mapping(session, neutron_id, nsx_switch_id, nsx_port_id): session.begin(subtransactions=True) try: mapping = models.NeutronNsxPortMapping( neutron_id, nsx_switch_id, nsx_port_id) session.add(mapping) session.commit() except db_exc.DBDuplicateEntry: with excutils.save_and_reraise_exception() as ctxt: session.rollback() # do not complain if the same exact mapping is being added, # otherwise re-raise because even though it is possible for the # same neutron port to map to different back-end ports over time, # this should not occur whilst a mapping already exists current = get_nsx_switch_and_port_id(session, neutron_id) if current[1] == nsx_port_id: LOG.debug(_("Port mapping for %s already available"), neutron_id) ctxt.reraise = False except db_exc.DBError: with excutils.save_and_reraise_exception(): # rollback for any other db error session.rollback() return mapping
**** CubicPower OpenStack Study **** def add_neutron_nsx_router_mapping(session, neutron_id, nsx_router_id): with session.begin(subtransactions=True): mapping = models.NeutronNsxRouterMapping( neutron_id=neutron_id, nsx_id=nsx_router_id) session.add(mapping) return mapping
**** CubicPower OpenStack Study **** def add_neutron_nsx_security_group_mapping(session, neutron_id, nsx_id): """Map a Neutron security group to a NSX security profile. :param session: a valid database session object :param neutron_id: a neutron security group identifier :param nsx_id: a nsx security profile identifier """ with session.begin(subtransactions=True): mapping = models.NeutronNsxSecurityGroupMapping( neutron_id=neutron_id, nsx_id=nsx_id) session.add(mapping) return mapping
**** CubicPower OpenStack Study **** def get_nsx_switch_ids(session, neutron_id): # This function returns a list of NSX switch identifiers because of # the possibility of chained logical switches return [mapping['nsx_id'] for mapping in session.query(models.NeutronNsxNetworkMapping).filter_by( neutron_id=neutron_id)]
**** CubicPower OpenStack Study **** def get_nsx_switch_and_port_id(session, neutron_id): try: mapping = (session.query(models.NeutronNsxPortMapping). filter_by(neutron_id=neutron_id). one()) return mapping['nsx_switch_id'], mapping['nsx_port_id'] except exc.NoResultFound: LOG.debug(_("NSX identifiers for neutron port %s not yet " "stored in Neutron DB"), neutron_id) return None, None
**** CubicPower OpenStack Study **** def get_nsx_router_id(session, neutron_id): try: mapping = (session.query(models.NeutronNsxRouterMapping). filter_by(neutron_id=neutron_id).one()) return mapping['nsx_id'] except exc.NoResultFound: LOG.debug(_("NSX identifiers for neutron router %s not yet " "stored in Neutron DB"), neutron_id)
**** CubicPower OpenStack Study **** def get_nsx_security_group_id(session, neutron_id): """Return the id of a security group in the NSX backend. Note: security groups are called 'security profiles' in NSX """ try: mapping = (session.query(models.NeutronNsxSecurityGroupMapping). filter_by(neutron_id=neutron_id). one()) return mapping['nsx_id'] except exc.NoResultFound: LOG.debug(_("NSX identifiers for neutron security group %s not yet " "stored in Neutron DB"), neutron_id) return None
**** CubicPower OpenStack Study **** def _delete_by_neutron_id(session, model, neutron_id): return session.query(model).filter_by(neutron_id=neutron_id).delete()
**** CubicPower OpenStack Study **** def delete_neutron_nsx_port_mapping(session, neutron_id): return _delete_by_neutron_id( session, models.NeutronNsxPortMapping, neutron_id)
**** CubicPower OpenStack Study **** def delete_neutron_nsx_router_mapping(session, neutron_id): return _delete_by_neutron_id( session, models.NeutronNsxRouterMapping, neutron_id)
**** CubicPower OpenStack Study **** def unset_default_network_gateways(session): with session.begin(subtransactions=True): session.query(networkgw_db.NetworkGateway).update( {networkgw_db.NetworkGateway.default: False})
**** CubicPower OpenStack Study **** def set_default_network_gateway(session, gw_id): with session.begin(subtransactions=True): gw = (session.query(networkgw_db.NetworkGateway). filter_by(id=gw_id).one()) gw['default'] = True
**** CubicPower OpenStack Study **** def set_multiprovider_network(session, network_id): with session.begin(subtransactions=True): multiprovider_network = models.MultiProviderNetworks( network_id) session.add(multiprovider_network) return multiprovider_network
**** CubicPower OpenStack Study **** def is_multiprovider_network(session, network_id): with session.begin(subtransactions=True): return bool( session.query(models.MultiProviderNetworks).filter_by( network_id=network_id).first())
|