<s> from oslo_db import exception as db_exc <EOL> from oslo_utils import uuidutils <EOL> from sqlalchemy . orm import exc as orm_exc <EOL> from neutron . ipam . drivers . neutrondb_ipam import db_models <EOL> from neutron . ipam import exceptions as ipam_exc <EOL> class IpamSubnetManager ( object ) : <EOL> @ classmethod <EOL> def load_by_neutron_subnet_id ( cls , session , neutron_subnet_id ) : <EOL> return session . query ( db_models . IpamSubnet ) . filter_by ( <EOL> neutron_subnet_id = neutron_subnet_id ) . first ( ) <EOL> def __init__ ( self , ipam_subnet_id , neutron_subnet_id ) : <EOL> self . _ipam_subnet_id = ipam_subnet_id <EOL> self . _neutron_subnet_id = neutron_subnet_id <EOL> @ property <EOL> def neutron_id ( self ) : <EOL> return self . _neutron_subnet_id <EOL> def create ( self , session ) : <EOL> """<STR_LIT>""" <EOL> if not self . _ipam_subnet_id : <EOL> self . _ipam_subnet_id = uuidutils . generate_uuid ( ) <EOL> ipam_subnet = db_models . IpamSubnet ( <EOL> id = self . _ipam_subnet_id , <EOL> neutron_subnet_id = self . _neutron_subnet_id ) <EOL> session . add ( ipam_subnet ) <EOL> return self . _ipam_subnet_id <EOL> @ classmethod <EOL> def delete ( cls , session , neutron_subnet_id ) : <EOL> """<STR_LIT>""" <EOL> return session . query ( db_models . IpamSubnet ) . filter_by ( <EOL> neutron_subnet_id = neutron_subnet_id ) . delete ( ) <EOL> def create_pool ( self , session , pool_start , pool_end ) : <EOL> """<STR_LIT>""" <EOL> ip_pool = db_models . IpamAllocationPool ( <EOL> ipam_subnet_id = self . _ipam_subnet_id , <EOL> first_ip = pool_start , <EOL> last_ip = pool_end ) <EOL> session . add ( ip_pool ) <EOL> ip_range = db_models . IpamAvailabilityRange ( <EOL> allocation_pool = ip_pool , <EOL> first_ip = pool_start , <EOL> last_ip = pool_end ) <EOL> session . add ( ip_range ) <EOL> return ip_pool <EOL> def delete_allocation_pools ( self , session ) : <EOL> """<STR_LIT>""" <EOL> session . query ( db_models . IpamAllocationPool ) . filter_by ( <EOL> ipam_subnet_id = self . _ipam_subnet_id ) . delete ( ) <EOL> def list_pools ( self , session ) : <EOL> """<STR_LIT>""" <EOL> return session . query ( <EOL> db_models . IpamAllocationPool ) . filter_by ( <EOL> ipam_subnet_id = self . _ipam_subnet_id ) <EOL> def _range_query ( self , session ) : <EOL> return session . query ( <EOL> db_models . IpamAvailabilityRange ) . join ( <EOL> db_models . IpamAllocationPool ) . filter_by ( <EOL> ipam_subnet_id = self . _ipam_subnet_id ) <EOL> def get_first_range ( self , session ) : <EOL> """<STR_LIT>""" <EOL> return self . _range_query ( session ) . first ( ) <EOL> def list_ranges_by_subnet_id ( self , session ) : <EOL> """<STR_LIT>""" <EOL> return self . _range_query ( session ) <EOL> def list_ranges_by_allocation_pool ( self , session , allocation_pool_id ) : <EOL> """<STR_LIT>""" <EOL> return session . query ( <EOL> db_models . IpamAvailabilityRange ) . join ( <EOL> db_models . IpamAllocationPool ) . filter_by ( <EOL> id = allocation_pool_id ) <EOL> def update_range ( self , session , db_range , first_ip = None , last_ip = None ) : <EOL> """<STR_LIT>""" <EOL> opts = { } <EOL> if first_ip : <EOL> opts [ '<STR_LIT>' ] = str ( first_ip ) <EOL> if last_ip : <EOL> opts [ '<STR_LIT>' ] = str ( last_ip ) <EOL> if not opts : <EOL> raise ipam_exc . IpamAvailabilityRangeNoChanges ( ) <EOL> try : <EOL> return session . query ( <EOL> db_models . IpamAvailabilityRange ) . filter_by ( <EOL> allocation_pool_id = db_range . allocation_pool_id ) . filter_by ( <EOL> first_ip = db_range . first_ip ) . filter_by ( <EOL> last_ip = db_range . last_ip ) . update ( opts ) <EOL> except orm_exc . ObjectDeletedError : <EOL> raise db_exc . RetryRequest ( ipam_exc . IPAllocationFailed ( ) ) <EOL> def delete_range ( self , session , db_range ) : <EOL> """<STR_LIT>""" <EOL> try : <EOL> return session . query ( <EOL> db_models . IpamAvailabilityRange ) . filter_by ( <EOL> allocation_pool_id = db_range . allocation_pool_id ) . filter_by ( <EOL> first_ip = db_range . first_ip ) . filter_by ( <EOL> last_ip = db_range . last_ip ) . delete ( ) <EOL> except orm_exc . ObjectDeletedError : <EOL> raise db_exc . RetryRequest ( ipam_exc . IPAllocationFailed ( ) ) <EOL> def create_range ( self , session , allocation_pool_id , <EOL> range_start , range_end ) : <EOL> """<STR_LIT>""" <EOL> new_ip_range = db_models . IpamAvailabilityRange ( <EOL> allocation_pool_id = allocation_pool_id , <EOL> first_ip = range_start , <EOL> last_ip = range_end ) <EOL> session . add ( new_ip_range ) <EOL> return new_ip_range <EOL> def check_unique_allocation ( self , session , ip_address ) : <EOL> """<STR_LIT>""" <EOL> iprequest = session . query ( db_models . IpamAllocation ) . filter_by ( <EOL> ipam_subnet_id = self . _ipam_subnet_id , status = '<STR_LIT>' , <EOL> ip_address = ip_address ) . first ( ) <EOL> if iprequest : <EOL> return False <EOL> return True <EOL> def list_allocations ( self , session , status = '<STR_LIT>' ) : <EOL> """<STR_LIT>""" <EOL> return session . query ( <EOL> db_models . IpamAllocation ) . filter_by ( <EOL> ipam_subnet_id = self . _ipam_subnet_id , <EOL> status = status ) <EOL> def create_allocation ( self , session , ip_address , <EOL> status = '<STR_LIT>' ) : <EOL> """<STR_LIT>""" <EOL> ip_request = db_models . IpamAllocation ( <EOL> ip_address = ip_address , <EOL> status = status , <EOL> ipam_subnet_id = self . _ipam_subnet_id ) <EOL> session . add ( ip_request ) <EOL> def delete_allocation ( self , session , ip_address ) : <EOL> """<STR_LIT>""" <EOL> return session . query ( db_models . IpamAllocation ) . filter_by ( <EOL> ip_address = ip_address , <EOL> ipam_subnet_id = self . _ipam_subnet_id ) . delete ( <EOL> synchronize_session = False ) </s>
