<s> from neutron_lib . api import validators <EOL> import sqlalchemy as sa <EOL> from sqlalchemy import orm <EOL> from neutron . api . v2 import attributes <EOL> from neutron . db import db_base_plugin_v2 <EOL> from neutron . db import model_base <EOL> from neutron . db import models_v2 <EOL> from neutron . db import portbindings_base <EOL> from neutron . extensions import portbindings <EOL> class PortBindingPort ( model_base . BASEV2 ) : <EOL> port_id = sa . Column ( sa . String ( <NUM_LIT> ) , <EOL> sa . ForeignKey ( '<STR_LIT>' , ondelete = "<STR_LIT>" ) , <EOL> primary_key = True ) <EOL> host = sa . Column ( sa . String ( <NUM_LIT:255> ) , nullable = False ) <EOL> port = orm . relationship ( <EOL> models_v2 . Port , <EOL> backref = orm . backref ( "<STR_LIT>" , <EOL> lazy = '<STR_LIT>' , uselist = False , <EOL> cascade = '<STR_LIT>' ) ) <EOL> class PortBindingMixin ( portbindings_base . PortBindingBaseMixin ) : <EOL> extra_binding_dict = None <EOL> def _port_model_hook ( self , context , original_model , query ) : <EOL> query = query . outerjoin ( PortBindingPort , <EOL> ( original_model . id == <EOL> PortBindingPort . port_id ) ) <EOL> return query <EOL> def _port_result_filter_hook ( self , query , filters ) : <EOL> values = filters and filters . get ( portbindings . HOST_ID , [ ] ) <EOL> if not values : <EOL> return query <EOL> query = query . filter ( PortBindingPort . host . in_ ( values ) ) <EOL> return query <EOL> db_base_plugin_v2 . NeutronDbPluginV2 . register_model_query_hook ( <EOL> models_v2 . Port , <EOL> "<STR_LIT>" , <EOL> '<STR_LIT>' , <EOL> None , <EOL> '<STR_LIT>' ) <EOL> def _process_portbindings_create_and_update ( self , context , port_data , <EOL> port ) : <EOL> binding_profile = port . get ( portbindings . PROFILE ) <EOL> binding_profile_set = validators . is_attr_set ( binding_profile ) <EOL> if not binding_profile_set and binding_profile is not None : <EOL> del port [ portbindings . PROFILE ] <EOL> binding_vnic = port . get ( portbindings . VNIC_TYPE ) <EOL> binding_vnic_set = validators . is_attr_set ( binding_vnic ) <EOL> if not binding_vnic_set and binding_vnic is not None : <EOL> del port [ portbindings . VNIC_TYPE ] <EOL> host = port_data . get ( portbindings . HOST_ID ) <EOL> host_set = validators . is_attr_set ( host ) <EOL> with context . session . begin ( subtransactions = True ) : <EOL> bind_port = context . session . query ( <EOL> PortBindingPort ) . filter_by ( port_id = port [ '<STR_LIT:id>' ] ) . first ( ) <EOL> if host_set : <EOL> if not bind_port : <EOL> context . session . add ( PortBindingPort ( port_id = port [ '<STR_LIT:id>' ] , <EOL> host = host ) ) <EOL> else : <EOL> bind_port . host = host <EOL> else : <EOL> host = bind_port . host if bind_port else None <EOL> self . _extend_port_dict_binding_host ( port , host ) <EOL> def get_port_host ( self , context , port_id ) : <EOL> with context . session . begin ( subtransactions = True ) : <EOL> bind_port = context . session . query ( <EOL> PortBindingPort ) . filter_by ( port_id = port_id ) . first ( ) <EOL> return bind_port . host if bind_port else None <EOL> def _extend_port_dict_binding_host ( self , port_res , host ) : <EOL> super ( PortBindingMixin , self ) . extend_port_dict_binding ( <EOL> port_res , None ) <EOL> port_res [ portbindings . HOST_ID ] = host <EOL> def extend_port_dict_binding ( self , port_res , port_db ) : <EOL> host = port_db . portbinding . host if port_db . portbinding else None <EOL> self . _extend_port_dict_binding_host ( port_res , host ) <EOL> def _extend_port_dict_binding ( plugin , port_res , port_db ) : <EOL> if not isinstance ( plugin , PortBindingMixin ) : <EOL> return <EOL> plugin . extend_port_dict_binding ( port_res , port_db ) <EOL> db_base_plugin_v2 . NeutronDbPluginV2 . register_dict_extend_funcs ( <EOL> attributes . PORTS , [ _extend_port_dict_binding ] ) </s>
