<s> import boto <EOL> from boto . connection import AWSQueryConnection <EOL> from boto . regioninfo import RegionInfo <EOL> from boto . exception import JSONResponseError <EOL> from boto . directconnect import exceptions <EOL> from boto . compat import json <EOL> class DirectConnectConnection ( AWSQueryConnection ) : <EOL> """<STR_LIT>""" <EOL> APIVersion = "<STR_LIT>" <EOL> DefaultRegionName = "<STR_LIT>" <EOL> DefaultRegionEndpoint = "<STR_LIT>" <EOL> ServiceName = "<STR_LIT>" <EOL> TargetPrefix = "<STR_LIT>" <EOL> ResponseError = JSONResponseError <EOL> _faults = { <EOL> "<STR_LIT>" : exceptions . DirectConnectClientException , <EOL> "<STR_LIT>" : exceptions . DirectConnectServerException , <EOL> } <EOL> def __init__ ( self , ** kwargs ) : <EOL> region = kwargs . pop ( '<STR_LIT>' , None ) <EOL> if not region : <EOL> region = RegionInfo ( self , self . DefaultRegionName , <EOL> self . DefaultRegionEndpoint ) <EOL> if '<STR_LIT:host>' not in kwargs : <EOL> kwargs [ '<STR_LIT:host>' ] = region . endpoint <EOL> super ( DirectConnectConnection , self ) . __init__ ( ** kwargs ) <EOL> self . region = region <EOL> def _required_auth_capability ( self ) : <EOL> return [ '<STR_LIT>' ] <EOL> def allocate_connection_on_interconnect ( self , bandwidth , connection_name , <EOL> owner_account , interconnect_id , <EOL> vlan ) : <EOL> """<STR_LIT>""" <EOL> params = { <EOL> '<STR_LIT>' : bandwidth , <EOL> '<STR_LIT>' : connection_name , <EOL> '<STR_LIT>' : owner_account , <EOL> '<STR_LIT>' : interconnect_id , <EOL> '<STR_LIT>' : vlan , <EOL> } <EOL> return self . make_request ( action = '<STR_LIT>' , <EOL> body = json . dumps ( params ) ) <EOL> def allocate_private_virtual_interface ( self , connection_id , <EOL> owner_account , <EOL> new_private_virtual_interface_allocation ) : <EOL> """<STR_LIT>""" <EOL> params = { <EOL> '<STR_LIT>' : connection_id , <EOL> '<STR_LIT>' : owner_account , <EOL> '<STR_LIT>' : new_private_virtual_interface_allocation , <EOL> } <EOL> return self . make_request ( action = '<STR_LIT>' , <EOL> body = json . dumps ( params ) ) <EOL> def allocate_public_virtual_interface ( self , connection_id , owner_account , <EOL> new_public_virtual_interface_allocation ) : <EOL> """<STR_LIT>""" <EOL> params = { <EOL> '<STR_LIT>' : connection_id , <EOL> '<STR_LIT>' : owner_account , <EOL> '<STR_LIT>' : new_public_virtual_interface_allocation , <EOL> } <EOL> return self . make_request ( action = '<STR_LIT>' , <EOL> body = json . dumps ( params ) ) <EOL> def confirm_connection ( self , connection_id ) : <EOL> """<STR_LIT>""" <EOL> params = { '<STR_LIT>' : connection_id , } <EOL> return self . make_request ( action = '<STR_LIT>' , <EOL> body = json . dumps ( params ) ) <EOL> def confirm_private_virtual_interface ( self , virtual_interface_id , <EOL> virtual_gateway_id ) : <EOL> """<STR_LIT>""" <EOL> params = { <EOL> '<STR_LIT>' : virtual_interface_id , <EOL> '<STR_LIT>' : virtual_gateway_id , <EOL> } <EOL> return self . make_request ( action = '<STR_LIT>' , <EOL> body = json . dumps ( params ) ) <EOL> def confirm_public_virtual_interface ( self , virtual_interface_id ) : <EOL> """<STR_LIT>""" <EOL> params = { '<STR_LIT>' : virtual_interface_id , } <EOL> return self . make_request ( action = '<STR_LIT>' , <EOL> body = json . dumps ( params ) ) <EOL> def create_connection ( self , location , bandwidth , connection_name ) : <EOL> """<STR_LIT>""" <EOL> params = { <EOL> '<STR_LIT:location>' : location , <EOL> '<STR_LIT>' : bandwidth , <EOL> '<STR_LIT>' : connection_name , <EOL> } <EOL> return self . make_request ( action = '<STR_LIT>' , <EOL> body = json . dumps ( params ) ) <EOL> def create_interconnect ( self , interconnect_name , bandwidth , location ) : <EOL> """<STR_LIT>""" <EOL> params = { <EOL> '<STR_LIT>' : interconnect_name , <EOL> '<STR_LIT>' : bandwidth , <EOL> '<STR_LIT:location>' : location , <EOL> } <EOL> return self . make_request ( action = '<STR_LIT>' , <EOL> body = json . dumps ( params ) ) <EOL> def create_private_virtual_interface ( self , connection_id , <EOL> new_private_virtual_interface ) : <EOL> """<STR_LIT>""" <EOL> params = { <EOL> '<STR_LIT>' : connection_id , <EOL> '<STR_LIT>' : new_private_virtual_interface , <EOL> } <EOL> return self . make_request ( action = '<STR_LIT>' , <EOL> body = json . dumps ( params ) ) <EOL> def create_public_virtual_interface ( self , connection_id , <EOL> new_public_virtual_interface ) : <EOL> """<STR_LIT>""" <EOL> params = { <EOL> '<STR_LIT>' : connection_id , <EOL> '<STR_LIT>' : new_public_virtual_interface , <EOL> } <EOL> return self . make_request ( action = '<STR_LIT>' , <EOL> body = json . dumps ( params ) ) <EOL> def delete_connection ( self , connection_id ) : <EOL> """<STR_LIT>""" <EOL> params = { '<STR_LIT>' : connection_id , } <EOL> return self . make_request ( action = '<STR_LIT>' , <EOL> body = json . dumps ( params ) ) <EOL> def delete_interconnect ( self , interconnect_id ) : <EOL> """<STR_LIT>""" <EOL> params = { '<STR_LIT>' : interconnect_id , } <EOL> return self . make_request ( action = '<STR_LIT>' , <EOL> body = json . dumps ( params ) ) <EOL> def delete_virtual_interface ( self , virtual_interface_id ) : <EOL> """<STR_LIT>""" <EOL> params = { '<STR_LIT>' : virtual_interface_id , } <EOL> return self . make_request ( action = '<STR_LIT>' , <EOL> body = json . dumps ( params ) ) <EOL> def describe_connections ( self , connection_id = None ) : <EOL> """<STR_LIT>""" <EOL> params = { } <EOL> if connection_id is not None : <EOL> params [ '<STR_LIT>' ] = connection_id <EOL> return self . make_request ( action = '<STR_LIT>' , <EOL> body = json . dumps ( params ) ) <EOL> def describe_connections_on_interconnect ( self , interconnect_id ) : <EOL> """<STR_LIT>""" <EOL> params = { '<STR_LIT>' : interconnect_id , } <EOL> return self . make_request ( action = '<STR_LIT>' , <EOL> body = json . dumps ( params ) ) <EOL> def describe_interconnects ( self , interconnect_id = None ) : <EOL> """<STR_LIT>""" <EOL> params = { } <EOL> if interconnect_id is not None : <EOL> params [ '<STR_LIT>' ] = interconnect_id <EOL> return self . make_request ( action = '<STR_LIT>' , <EOL> body = json . dumps ( params ) ) <EOL> def describe_locations ( self ) : <EOL> """<STR_LIT>""" <EOL> params = { } <EOL> return self . make_request ( action = '<STR_LIT>' , <EOL> body = json . dumps ( params ) ) <EOL> def describe_virtual_gateways ( self ) : <EOL> """<STR_LIT>""" <EOL> params = { } <EOL> return self . make_request ( action = '<STR_LIT>' , <EOL> body = json . dumps ( params ) ) <EOL> def describe_virtual_interfaces ( self , connection_id = None , <EOL> virtual_interface_id = None ) : <EOL> """<STR_LIT>""" <EOL> params = { } <EOL> if connection_id is not None : <EOL> params [ '<STR_LIT>' ] = connection_id <EOL> if virtual_interface_id is not None : <EOL> params [ '<STR_LIT>' ] = virtual_interface_id <EOL> return self . make_request ( action = '<STR_LIT>' , <EOL> body = json . dumps ( params ) ) <EOL> def make_request ( self , action , body ) : <EOL> headers = { <EOL> '<STR_LIT>' : '<STR_LIT>' % ( self . TargetPrefix , action ) , <EOL> '<STR_LIT>' : self . region . endpoint , <EOL> '<STR_LIT:Content-Type>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : str ( len ( body ) ) , <EOL> } <EOL> http_request = self . build_base_http_request ( <EOL> method = '<STR_LIT:POST>' , path = '<STR_LIT:/>' , auth_path = '<STR_LIT:/>' , params = { } , <EOL> headers = headers , data = body ) <EOL> response = self . _mexe ( http_request , sender = None , <EOL> override_num_retries = <NUM_LIT:10> ) <EOL> response_body = response . read ( ) . decode ( '<STR_LIT:utf-8>' ) <EOL> boto . log . debug ( response_body ) <EOL> if response . status == <NUM_LIT:200> : <EOL> if response_body : <EOL> return json . loads ( response_body ) <EOL> else : <EOL> json_body = json . loads ( response_body ) <EOL> fault_name = json_body . get ( '<STR_LIT>' , None ) <EOL> exception_class = self . _faults . get ( fault_name , self . ResponseError ) <EOL> raise exception_class ( response . status , response . reason , <EOL> body = json_body ) </s>
