<s> from sqlalchemy import func <EOL> from sqlalchemy import sql <EOL> from neutron . db import agents_db <EOL> from neutron . db import l3_agentschedulers_db as l3_sch_db <EOL> from neutron . db import l3_attrs_db <EOL> from neutron . db import l3_db <EOL> class L3_HA_scheduler_db_mixin ( l3_sch_db . AZL3AgentSchedulerDbMixin ) : <EOL> def get_ha_routers_l3_agents_count ( self , context ) : <EOL> """<STR_LIT>""" <EOL> binding_model = l3_sch_db . RouterL3AgentBinding <EOL> sub_query = ( context . session . query ( <EOL> binding_model . router_id , <EOL> func . count ( binding_model . router_id ) . label ( '<STR_LIT:count>' ) ) . <EOL> join ( l3_attrs_db . RouterExtraAttributes , <EOL> binding_model . router_id == <EOL> l3_attrs_db . RouterExtraAttributes . router_id ) . <EOL> join ( l3_db . Router ) . <EOL> filter ( l3_attrs_db . RouterExtraAttributes . ha == sql . true ( ) ) . <EOL> group_by ( binding_model . router_id ) . subquery ( ) ) <EOL> query = ( context . session . query ( l3_db . Router , sub_query . c . count ) . <EOL> join ( sub_query ) ) <EOL> return [ ( self . _make_router_dict ( router ) , agent_count ) <EOL> for router , agent_count in query ] <EOL> def get_l3_agents_ordered_by_num_routers ( self , context , agent_ids ) : <EOL> if not agent_ids : <EOL> return [ ] <EOL> query = ( context . session . query ( agents_db . Agent , func . count ( <EOL> l3_sch_db . RouterL3AgentBinding . router_id ) . label ( '<STR_LIT:count>' ) ) . <EOL> outerjoin ( l3_sch_db . RouterL3AgentBinding ) . <EOL> group_by ( agents_db . Agent . id ) . <EOL> filter ( agents_db . Agent . id . in_ ( agent_ids ) ) . <EOL> order_by ( '<STR_LIT:count>' ) ) <EOL> return [ record [ <NUM_LIT:0> ] for record in query ] <EOL> def _get_agents_dict_for_router ( self , agents_and_states ) : <EOL> agents = [ ] <EOL> for agent , ha_state in agents_and_states : <EOL> l3_agent_dict = self . _make_agent_dict ( agent ) <EOL> l3_agent_dict [ '<STR_LIT>' ] = ha_state <EOL> agents . append ( l3_agent_dict ) <EOL> return { '<STR_LIT>' : agents } <EOL> def list_l3_agents_hosting_router ( self , context , router_id ) : <EOL> with context . session . begin ( subtransactions = True ) : <EOL> router_db = self . _get_router ( context , router_id ) <EOL> if router_db . extra_attributes . ha : <EOL> bindings = self . get_l3_bindings_hosting_router_with_ha_states ( <EOL> context , router_id ) <EOL> else : <EOL> bindings = self . _get_l3_bindings_hosting_routers ( <EOL> context , [ router_id ] ) <EOL> bindings = [ ( binding . l3_agent , None ) for binding in bindings ] <EOL> return self . _get_agents_dict_for_router ( bindings ) </s>
