<s> from neutron_lib import constants <EOL> from oslo_utils import uuidutils <EOL> import sqlalchemy as sa <EOL> from sqlalchemy . orm import exc <EOL> from neutron . _i18n import _ <EOL> from neutron . api . v2 import attributes as attr <EOL> from neutron . db import db_base_plugin_v2 <EOL> from neutron . db import model_base <EOL> from neutron . extensions import address_scope as ext_address_scope <EOL> class AddressScope ( model_base . BASEV2 , model_base . HasId , model_base . HasTenant ) : <EOL> """<STR_LIT>""" <EOL> __tablename__ = "<STR_LIT>" <EOL> name = sa . Column ( sa . String ( attr . NAME_MAX_LEN ) , nullable = False ) <EOL> shared = sa . Column ( sa . Boolean , nullable = False ) <EOL> ip_version = sa . Column ( sa . Integer ( ) , nullable = False ) <EOL> class AddressScopeDbMixin ( ext_address_scope . AddressScopePluginBase ) : <EOL> """<STR_LIT>""" <EOL> __native_bulk_support = True <EOL> def _make_address_scope_dict ( self , address_scope , fields = None ) : <EOL> res = { '<STR_LIT:id>' : address_scope [ '<STR_LIT:id>' ] , <EOL> '<STR_LIT:name>' : address_scope [ '<STR_LIT:name>' ] , <EOL> '<STR_LIT>' : address_scope [ '<STR_LIT>' ] , <EOL> '<STR_LIT>' : address_scope [ '<STR_LIT>' ] , <EOL> '<STR_LIT>' : address_scope [ '<STR_LIT>' ] } <EOL> return self . _fields ( res , fields ) <EOL> def _get_address_scope ( self , context , id ) : <EOL> try : <EOL> return self . _get_by_id ( context , AddressScope , id ) <EOL> except exc . NoResultFound : <EOL> raise ext_address_scope . AddressScopeNotFound ( address_scope_id = id ) <EOL> def is_address_scope_owned_by_tenant ( self , context , id ) : <EOL> """<STR_LIT>""" <EOL> address_scope = self . _get_address_scope ( context , id ) <EOL> return context . is_admin or ( <EOL> address_scope . tenant_id == context . tenant_id ) <EOL> def get_ip_version_for_address_scope ( self , context , id ) : <EOL> address_scope = self . _get_address_scope ( context , id ) <EOL> return address_scope . ip_version <EOL> def create_address_scope ( self , context , address_scope ) : <EOL> """<STR_LIT>""" <EOL> a_s = address_scope [ '<STR_LIT>' ] <EOL> address_scope_id = a_s . get ( '<STR_LIT:id>' ) or uuidutils . generate_uuid ( ) <EOL> with context . session . begin ( subtransactions = True ) : <EOL> pool_args = { '<STR_LIT>' : a_s [ '<STR_LIT>' ] , <EOL> '<STR_LIT:id>' : address_scope_id , <EOL> '<STR_LIT:name>' : a_s [ '<STR_LIT:name>' ] , <EOL> '<STR_LIT>' : a_s [ '<STR_LIT>' ] , <EOL> '<STR_LIT>' : a_s [ '<STR_LIT>' ] } <EOL> address_scope = AddressScope ( ** pool_args ) <EOL> context . session . add ( address_scope ) <EOL> return self . _make_address_scope_dict ( address_scope ) <EOL> def update_address_scope ( self , context , id , address_scope ) : <EOL> a_s = address_scope [ '<STR_LIT>' ] <EOL> with context . session . begin ( subtransactions = True ) : <EOL> address_scope = self . _get_address_scope ( context , id ) <EOL> if address_scope . shared and not a_s . get ( '<STR_LIT>' , True ) : <EOL> reason = _ ( "<STR_LIT>" ) <EOL> raise ext_address_scope . AddressScopeUpdateError ( <EOL> address_scope_id = id , reason = reason ) <EOL> address_scope . update ( a_s ) <EOL> return self . _make_address_scope_dict ( address_scope ) <EOL> def get_address_scope ( self , context , id , fields = None ) : <EOL> address_scope = self . _get_address_scope ( context , id ) <EOL> return self . _make_address_scope_dict ( address_scope , fields ) <EOL> def get_address_scopes ( self , context , filters = None , fields = None , <EOL> sorts = None , limit = None , marker = None , <EOL> page_reverse = False ) : <EOL> marker_obj = self . _get_marker_obj ( context , '<STR_LIT>' , limit , marker ) <EOL> collection = self . _get_collection ( context , AddressScope , <EOL> self . _make_address_scope_dict , <EOL> filters = filters , fields = fields , <EOL> sorts = sorts , <EOL> limit = limit , <EOL> marker_obj = marker_obj , <EOL> page_reverse = page_reverse ) <EOL> return collection <EOL> def get_address_scopes_count ( self , context , filters = None ) : <EOL> return self . _get_collection_count ( context , AddressScope , <EOL> filters = filters ) <EOL> def delete_address_scope ( self , context , id ) : <EOL> with context . session . begin ( subtransactions = True ) : <EOL> if self . _get_subnetpools_by_address_scope_id ( context , id ) : <EOL> raise ext_address_scope . AddressScopeInUse ( address_scope_id = id ) <EOL> address_scope = self . _get_address_scope ( context , id ) <EOL> context . session . delete ( address_scope ) <EOL> def _extend_network_dict_address_scope ( self , network_res , network_db ) : <EOL> network_res [ ext_address_scope . IPV4_ADDRESS_SCOPE ] = None <EOL> network_res [ ext_address_scope . IPV6_ADDRESS_SCOPE ] = None <EOL> subnetpools = { subnet . subnetpool for subnet in network_db . subnets <EOL> if subnet . subnetpool } <EOL> for subnetpool in subnetpools : <EOL> as_id = subnetpool [ ext_address_scope . ADDRESS_SCOPE_ID ] <EOL> if subnetpool [ '<STR_LIT>' ] == constants . IP_VERSION_4 : <EOL> network_res [ ext_address_scope . IPV4_ADDRESS_SCOPE ] = as_id <EOL> if subnetpool [ '<STR_LIT>' ] == constants . IP_VERSION_6 : <EOL> network_res [ ext_address_scope . IPV6_ADDRESS_SCOPE ] = as_id <EOL> return network_res <EOL> db_base_plugin_v2 . NeutronDbPluginV2 . register_dict_extend_funcs ( <EOL> attr . NETWORKS , [ '<STR_LIT>' ] ) </s>
