本文整理汇总了Python中quantum.db.api.get_session函数的典型用法代码示例。如果您正苦于以下问题:Python get_session函数的具体用法?Python get_session怎么用?Python get_session使用的例子?那么恭喜您, 这里精选的函数代码示例或许可以为您提供帮助。
在下文中一共展示了get_session函数的20个代码示例,这些例子默认根据受欢迎程度排序。您可以为喜欢或者感觉有用的代码点赞,您的评价将有助于我们的系统推荐出更棒的Python代码示例。
示例1: test_vlan_id_pool
def test_vlan_id_pool(self):
session = db.get_session()
vlan_ids = set()
for x in xrange(VLAN_MIN, VLAN_MAX + 1):
vlan_id = ovs_db_v2.reserve_vlan_id(db.get_session())
self.assertGreaterEqual(vlan_id, VLAN_MIN)
self.assertLessEqual(vlan_id, VLAN_MAX)
vlan_ids.add(vlan_id)
with self.assertRaises(q_exc.NoNetworkAvailable):
vlan_id = ovs_db_v2.reserve_vlan_id(session)
for vlan_id in vlan_ids:
ovs_db_v2.release_vlan_id(vlan_id)
开发者ID:missall,项目名称:quantum,代码行数:14,代码来源:test_ovs_db.py
示例2: ofp_has_servers
def ofp_has_servers():
session = db.get_session()
try:
session.query(models.OFPServers).first()
return True
except exc.NoResultFound:
return False
开发者ID:ykaneko,项目名称:quantum-backup,代码行数:7,代码来源:api.py
示例3: get_plugged_port
def get_plugged_port(interface_id):
session = db.get_session()
try:
return session.query(models.Port).\
filter_by(interface_id=interface_id).one()
except exc.NoResultFound:
return None
开发者ID:yasuhito,项目名称:quantum-nec-of-plugin,代码行数:7,代码来源:quantum_db_extension.py
示例4: get_port_properties
def get_port_properties(portid):
session = db.get_session()
try:
port = session.query(neuca_models.port_properties).filter_by(port_id=portid).one()
except exc.NoResultFound:
pass
return port
开发者ID:RENCI-NRIG,项目名称:neuca-agent,代码行数:7,代码来源:neuca_db.py
示例5: sync_vlan_allocations
def sync_vlan_allocations(self, network_vlan_ranges):
"""Synchronize vlan_allocations table with configured VLAN ranges."""
session = db_api.get_session()
with session.begin():
# get existing allocations for all physical networks
allocations = dict()
allocs_q = session.query(hyperv_model.VlanAllocation)
for alloc in allocs_q:
allocations.setdefault(alloc.physical_network,
set()).add(alloc)
# process vlan ranges for each configured physical network
for physical_network, vlan_ranges in network_vlan_ranges.items():
# determine current configured allocatable vlans for this
# physical network
vlan_ids = set()
for vlan_range in vlan_ranges:
vlan_ids |= set(xrange(vlan_range[0], vlan_range[1] + 1))
# remove from table unallocated vlans not currently allocatable
self._remove_non_allocatable_vlans(session,
physical_network,
vlan_ids,
allocations)
# add missing allocatable vlans to table
self._add_missing_allocatable_vlans(session, vlan_ids,
physical_network)
# remove from table unallocated vlans for any unconfigured physical
# networks
self._remove_unconfigured_vlans(session, allocations)
开发者ID:Apsu,项目名称:quantum,代码行数:33,代码来源:db.py
示例6: test_key_allocation
def test_key_allocation(self):
tunnel_key = db_api_v2.TunnelKey()
session = db.get_session()
with nested(self.network('network-0'),
self.network('network-1')
) as (network_0,
network_1):
network_id0 = network_0['network']['id']
key0 = tunnel_key.allocate(session, network_id0)
network_id1 = network_1['network']['id']
key1 = tunnel_key.allocate(session, network_id1)
key_list = tunnel_key.all_list()
self.assertEqual(len(key_list), 2)
expected_list = [(network_id0, key0), (network_id1, key1)]
self.assertEqual(self._tunnel_key_sort(key_list),
expected_list)
tunnel_key.delete(session, network_id0)
key_list = tunnel_key.all_list()
self.assertEqual(self._tunnel_key_sort(key_list),
[(network_id1, key1)])
tunnel_key.delete(session, network_id1)
self.assertEqual(tunnel_key.all_list(), [])
开发者ID:bbrahmbhatt,项目名称:quantum,代码行数:25,代码来源:test_ryu_db.py
示例7: get_device_details
def get_device_details(self, rpc_context, **kwargs):
"""Agent requests device details."""
agent_id = kwargs.get('agent_id')
device = kwargs.get('device')
LOG.debug(_("Device %(device)s details requested from %(agent_id)s"),
{'device': device, 'agent_id': agent_id})
port = self.get_port_from_device(device)
if port:
binding = db.get_network_binding(db_api.get_session(),
port['network_id'])
(network_type,
segmentation_id) = constants.interpret_vlan_id(binding.vlan_id)
entry = {'device': device,
'network_type': network_type,
'physical_network': binding.physical_network,
'segmentation_id': segmentation_id,
'network_id': port['network_id'],
'port_id': port['id'],
'admin_state_up': port['admin_state_up']}
if cfg.CONF.AGENT.rpc_support_old_agents:
entry['vlan_id'] = binding.vlan_id
new_status = (q_const.PORT_STATUS_ACTIVE if port['admin_state_up']
else q_const.PORT_STATUS_DOWN)
if port['status'] != new_status:
db.set_port_status(port['id'], new_status)
else:
entry = {'device': device}
LOG.debug(_("%s can not be found in database"), device)
return entry
开发者ID:citrix-openstack,项目名称:neutron,代码行数:29,代码来源:lb_quantum_plugin.py
示例8: update_vlan_id_pool
def update_vlan_id_pool():
"""Update vlan_ids based on current configuration."""
# determine current dynamically-allocated range
vlans = set(xrange(cfg.CONF.OVS.vlan_min,
cfg.CONF.OVS.vlan_max + 1))
session = db.get_session()
with session.begin(subtransactions=True):
# remove unused vlan_ids outside current range
try:
records = (session.query(ovs_models_v2.VlanID).
all())
for record in records:
try:
vlans.remove(record.vlan_id)
except KeyError:
if not record.vlan_used:
LOG.debug("removing vlan %s from pool"
% record.vlan_id)
session.delete(record)
except exc.NoResultFound:
pass
# add missing vlan_ids
for vlan in vlans:
record = ovs_models_v2.VlanID(vlan)
session.add(record)
开发者ID:vbannai,项目名称:quantum,代码行数:28,代码来源:ovs_db_v2.py
示例9: update_portbinding
def update_portbinding(port_id, blade_intf_dn=None, portprofile_name=None,
vlan_name=None, vlan_id=None, qos=None,
tenant_id=None, instance_id=None,
vif_id=None):
"""Updates port binding"""
LOG.debug("db update_portbinding() called")
session = db.get_session()
try:
port_binding = (session.query(ucs_models.PortBinding).
filter_by(port_id=port_id).one())
if blade_intf_dn:
port_binding.blade_intf_dn = blade_intf_dn
if portprofile_name:
port_binding.portprofile_name = portprofile_name
if vlan_name:
port_binding.vlan_name = vlan_name
if vlan_name:
port_binding.vlan_id = vlan_id
if qos:
port_binding.qos = qos
if tenant_id:
port_binding.tenant_id = tenant_id
if instance_id:
port_binding.instance_id = instance_id
if vif_id:
port_binding.vif_id = vif_id
session.merge(port_binding)
session.flush()
return port_binding
except exc.NoResultFound:
raise c_exc.PortVnicNotFound(port_id=port_id)
开发者ID:kumarcv,项目名称:openstack-nf,代码行数:31,代码来源:ucs_db_v2.py
示例10: get_port
def get_port(port_id):
session = db.get_session()
try:
port = session.query(models_v2.Port).filter_by(id=port_id).one()
except exc.NoResultFound:
port = None
return port
开发者ID:ewindisch,项目名称:quantum,代码行数:7,代码来源:ovs_db_v2.py
示例11: port_binding_get
def port_binding_get(port_id, net_id):
session = db.get_session()
session.query(models_v2.Port).filter(
models_v2.Port.network_id == net_id).filter(
models_v2.Port.id == port_id).one() # confirm port exists
return session.query(ryu_models_v2.PortBinding).filter_by(
network_id=net_id).filter_by(port_id=port_id).one()
开发者ID:brosenberg,项目名称:quantum-pub,代码行数:7,代码来源:api_v2.py
示例12: create_vlanids
def create_vlanids():
"""Prepopulates the vlan_bindings table"""
LOG.debug("create_vlanids() called")
session = db.get_session()
start = CONF.VLANS.vlan_start
end = CONF.VLANS.vlan_end
try:
vlanid = session.query(L2_MODEL.VlanID).one()
except exc.MultipleResultsFound:
"""
TODO (Sumit): Salvatore rightly points out that this will not handle
change in VLAN ID range across server reboots. This is currently not
a supported feature. This logic will need to change if this feature
has to be supported.
Per Dan's suggestion we just throw a server exception for now.
"""
current_start = (
int(session.query(func.min(L2_MODEL.VlanID.vlan_id)).
one()[0]))
current_end = (
int(session.query(func.max(L2_MODEL.VlanID.vlan_id)).
one()[0]))
if current_start != start or current_end != end:
LOG.debug("Old VLAN range %s-%s" % (current_start, current_end))
LOG.debug("New VLAN range %s-%s" % (start, end))
raise c_exc.UnableToChangeVlanRange(range_start=current_start,
range_end=current_end)
except exc.NoResultFound:
LOG.debug("Setting VLAN range to %s-%s" % (start, end))
while start <= end:
vlanid = L2_MODEL.VlanID(start)
session.add(vlanid)
start += 1
session.flush()
return
开发者ID:danhan,项目名称:quantum,代码行数:35,代码来源:l2network_db.py
示例13: set_ofp_servers
def set_ofp_servers(hosts):
session = db.get_session()
session.query(models.OFPServer).delete()
for (host_address, host_type) in hosts:
host = models.OFPServer(host_address, host_type)
session.add(host)
session.flush()
开发者ID:AnyBucket,项目名称:OpenStack-Install-and-Understand-Guide,代码行数:7,代码来源:api.py
示例14: test_invalid_specific_vlan_id
def test_invalid_specific_vlan_id(self):
session = db.get_session()
with self.assertRaises(q_exc.InvalidInput):
vlan_id = ovs_db_v2.reserve_specific_vlan_id(0, session)
with self.assertRaises(q_exc.InvalidInput):
vlan_id = ovs_db_v2.reserve_specific_vlan_id(4095, session)
开发者ID:missall,项目名称:quantum,代码行数:7,代码来源:test_ovs_db.py
示例15: reserve_vlanid
def reserve_vlanid():
"""Reserves the first unused vlanid"""
LOG.debug("reserve_vlanid() called")
session = db.get_session()
try:
rvlan = (session.query(L2_MODEL.VlanID).
first())
if not rvlan:
create_vlanids()
rvlan = (session.query(L2_MODEL.VlanID).
filter_by(vlan_used=False).
first())
if not rvlan:
raise c_exc.VlanIDNotAvailable()
rvlanid = (session.query(L2_MODEL.VlanID).
filter_by(vlan_id=rvlan["vlan_id"]).
one())
rvlanid["vlan_used"] = True
session.merge(rvlanid)
session.flush()
return rvlan["vlan_id"]
except exc.NoResultFound:
raise c_exc.VlanIDNotAvailable()
开发者ID:danhan,项目名称:quantum,代码行数:25,代码来源:l2network_db.py
示例16: get_port_from_device
def get_port_from_device(device):
"""Get port from database."""
LOG.debug(_("get_port_from_device() called"))
session = db.get_session()
sg_binding_port = sg_db.SecurityGroupPortBinding.port_id
query = session.query(models_v2.Port,
sg_db.SecurityGroupPortBinding.security_group_id)
query = query.outerjoin(sg_db.SecurityGroupPortBinding,
models_v2.Port.id == sg_binding_port)
query = query.filter(models_v2.Port.id.startswith(device))
port_and_sgs = query.all()
if not port_and_sgs:
return
port = port_and_sgs[0][0]
plugin = manager.QuantumManager.get_plugin()
port_dict = plugin._make_port_dict(port)
port_dict['security_groups'] = []
for port_in_db, sg_id in port_and_sgs:
if sg_id:
port_dict['security_groups'].append(sg_id)
port_dict['security_group_rules'] = []
port_dict['security_group_source_groups'] = []
port_dict['fixed_ips'] = [ip['ip_address']
for ip in port['fixed_ips']]
return port_dict
开发者ID:SangBaisong,项目名称:quantum,代码行数:26,代码来源:l2network_db_v2.py
示例17: test_ofp_server
def test_ofp_server(self):
session = db.get_session()
servers = session.query(ryu_models_v2.OFPServer).all()
print servers
self.assertEqual(len(servers), 2)
for s in servers:
self.assertTrue((s.address, s.host_type) in self.hosts)
开发者ID:nitishb,项目名称:quantum,代码行数:7,代码来源:test_ryu_db.py
示例18: sync_network_states
def sync_network_states(network_vlan_ranges):
"""Synchronize network_states table with current configured VLAN ranges."""
session = db.get_session()
with session.begin():
# get existing allocations for all physical networks
allocations = dict()
entries = session.query(mlnx_models_v2.SegmentationIdAllocation).all()
for entry in entries:
allocations.setdefault(entry.physical_network, set()).add(entry)
# process vlan ranges for each configured physical network
for physical_network, vlan_ranges in network_vlan_ranges.iteritems():
# determine current configured allocatable vlans for this
# physical network
vlan_ids = set()
for vlan_range in vlan_ranges:
vlan_ids |= set(xrange(vlan_range[0], vlan_range[1] + 1))
# remove from table unallocated vlans not currently allocatable
_remove_non_allocatable_vlans(session, allocations, physical_network, vlan_ids)
# add missing allocatable vlans to table
_add_missing_allocatable_vlans(session, physical_network, vlan_ids)
# remove from table unallocated vlans for any unconfigured physical
# networks
_remove_unconfigured_vlans(session, allocations)
开发者ID:soheilhy,项目名称:quantum,代码行数:28,代码来源:mlnx_db_v2.py
示例19: get_device_details
def get_device_details(self, rpc_context, **kwargs):
"""Agent requests device details"""
agent_id = kwargs.get('agent_id')
device = kwargs.get('device')
LOG.debug(_("Device %(device)s details requested from %(agent_id)s"),
locals())
port = self.get_port_from_device(device)
if port:
binding = db.get_network_binding(db_api.get_session(),
port['network_id'])
entry = {'device': device,
'physical_network': binding.physical_network,
'network_type': binding.network_type,
'vlan_id': binding.vlan_id,
'network_id': port['network_id'],
'port_id': port['id'],
'admin_state_up': port['admin_state_up']}
new_status = (q_const.PORT_STATUS_ACTIVE if port['admin_state_up']
else q_const.PORT_STATUS_DOWN)
if port['status'] != new_status:
db.set_port_status(port['id'], new_status)
else:
entry = {'device': device}
LOG.debug(_("%s can not be found in database"), device)
return entry
开发者ID:tpaszkowski,项目名称:quantum,代码行数:25,代码来源:lb_quantum_plugin.py
示例20: sync_tunnel_allocations
def sync_tunnel_allocations(tunnel_id_ranges):
"""Synchronize tunnel_allocations table with configured tunnel ranges"""
# determine current configured allocatable tunnels
tunnel_ids = set()
for tunnel_id_range in tunnel_id_ranges:
tun_min, tun_max = tunnel_id_range
if tun_max + 1 - tun_min > 1000000:
LOG.error(_("Skipping unreasonable tunnel ID range "
"%(tun_min)s:%(tun_max)s"),
locals())
else:
tunnel_ids |= set(xrange(tun_min, tun_max + 1))
session = db.get_session()
with session.begin():
# remove from table unallocated tunnels not currently allocatable
allocs = (session.query(ovs_models_v2.TunnelAllocation).
all())
for alloc in allocs:
try:
# see if tunnel is allocatable
tunnel_ids.remove(alloc.tunnel_id)
except KeyError:
# it's not allocatable, so check if its allocated
if not alloc.allocated:
# it's not, so remove it from table
LOG.debug(_("Removing tunnel %s from pool"),
alloc.tunnel_id)
session.delete(alloc)
# add missing allocatable tunnels to table
for tunnel_id in sorted(tunnel_ids):
alloc = ovs_models_v2.TunnelAllocation(tunnel_id)
session.add(alloc)
开发者ID:NCU-PDCLAB,项目名称:quantum,代码行数:35,代码来源:ovs_db_v2.py
注:本文中的quantum.db.api.get_session函数示例由纯净天空整理自Github/MSDocs等源码及文档管理平台,相关代码片段筛选自各路编程大神贡献的开源项目,源码版权归原作者所有,传播和使用请参考对应项目的License;未经允许,请勿转载。 |
请发表评论