<s> import boto <EOL> from boto . compat import json <EOL> from boto . connection import AWSQueryConnection <EOL> from boto . regioninfo import RegionInfo <EOL> from boto . exception import JSONResponseError <EOL> from boto . configservice import exceptions <EOL> class ConfigServiceConnection ( 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 . InvalidLimitException , <EOL> "<STR_LIT>" : exceptions . NoSuchBucketException , <EOL> "<STR_LIT>" : exceptions . InvalidSNSTopicARNException , <EOL> "<STR_LIT>" : exceptions . ResourceNotDiscoveredException , <EOL> "<STR_LIT>" : exceptions . MaxNumberOfDeliveryChannelsExceededException , <EOL> "<STR_LIT>" : exceptions . LastDeliveryChannelDeleteFailedException , <EOL> "<STR_LIT>" : exceptions . InsufficientDeliveryPolicyException , <EOL> "<STR_LIT>" : exceptions . InvalidRoleException , <EOL> "<STR_LIT>" : exceptions . InvalidTimeRangeException , <EOL> "<STR_LIT>" : exceptions . NoSuchDeliveryChannelException , <EOL> "<STR_LIT>" : exceptions . NoSuchConfigurationRecorderException , <EOL> "<STR_LIT>" : exceptions . InvalidS3KeyPrefixException , <EOL> "<STR_LIT>" : exceptions . InvalidDeliveryChannelNameException , <EOL> "<STR_LIT>" : exceptions . NoRunningConfigurationRecorderException , <EOL> "<STR_LIT>" : exceptions . ValidationException , <EOL> "<STR_LIT>" : exceptions . NoAvailableConfigurationRecorderException , <EOL> "<STR_LIT>" : exceptions . InvalidNextTokenException , <EOL> "<STR_LIT>" : exceptions . InvalidConfigurationRecorderNameException , <EOL> "<STR_LIT>" : exceptions . NoAvailableDeliveryChannelException , <EOL> "<STR_LIT>" : exceptions . MaxNumberOfConfigurationRecordersExceededException , <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 or kwargs [ '<STR_LIT:host>' ] is None : <EOL> kwargs [ '<STR_LIT:host>' ] = region . endpoint <EOL> super ( ConfigServiceConnection , self ) . __init__ ( ** kwargs ) <EOL> self . region = region <EOL> def _required_auth_capability ( self ) : <EOL> return [ '<STR_LIT>' ] <EOL> def delete_delivery_channel ( self , delivery_channel_name ) : <EOL> """<STR_LIT>""" <EOL> params = { '<STR_LIT>' : delivery_channel_name , } <EOL> return self . make_request ( action = '<STR_LIT>' , <EOL> body = json . dumps ( params ) ) <EOL> def deliver_config_snapshot ( self , delivery_channel_name ) : <EOL> """<STR_LIT>""" <EOL> params = { '<STR_LIT>' : delivery_channel_name , } <EOL> return self . make_request ( action = '<STR_LIT>' , <EOL> body = json . dumps ( params ) ) <EOL> def describe_configuration_recorder_status ( self , <EOL> configuration_recorder_names = None ) : <EOL> """<STR_LIT>""" <EOL> params = { } <EOL> if configuration_recorder_names is not None : <EOL> params [ '<STR_LIT>' ] = configuration_recorder_names <EOL> return self . make_request ( action = '<STR_LIT>' , <EOL> body = json . dumps ( params ) ) <EOL> def describe_configuration_recorders ( self , <EOL> configuration_recorder_names = None ) : <EOL> """<STR_LIT>""" <EOL> params = { } <EOL> if configuration_recorder_names is not None : <EOL> params [ '<STR_LIT>' ] = configuration_recorder_names <EOL> return self . make_request ( action = '<STR_LIT>' , <EOL> body = json . dumps ( params ) ) <EOL> def describe_delivery_channel_status ( self , delivery_channel_names = None ) : <EOL> """<STR_LIT>""" <EOL> params = { } <EOL> if delivery_channel_names is not None : <EOL> params [ '<STR_LIT>' ] = delivery_channel_names <EOL> return self . make_request ( action = '<STR_LIT>' , <EOL> body = json . dumps ( params ) ) <EOL> def describe_delivery_channels ( self , delivery_channel_names = None ) : <EOL> """<STR_LIT>""" <EOL> params = { } <EOL> if delivery_channel_names is not None : <EOL> params [ '<STR_LIT>' ] = delivery_channel_names <EOL> return self . make_request ( action = '<STR_LIT>' , <EOL> body = json . dumps ( params ) ) <EOL> def get_resource_config_history ( self , resource_type , resource_id , <EOL> later_time = None , earlier_time = None , <EOL> chronological_order = None , limit = None , <EOL> next_token = None ) : <EOL> """<STR_LIT>""" <EOL> params = { <EOL> '<STR_LIT>' : resource_type , <EOL> '<STR_LIT>' : resource_id , <EOL> } <EOL> if later_time is not None : <EOL> params [ '<STR_LIT>' ] = later_time <EOL> if earlier_time is not None : <EOL> params [ '<STR_LIT>' ] = earlier_time <EOL> if chronological_order is not None : <EOL> params [ '<STR_LIT>' ] = chronological_order <EOL> if limit is not None : <EOL> params [ '<STR_LIT>' ] = limit <EOL> if next_token is not None : <EOL> params [ '<STR_LIT>' ] = next_token <EOL> return self . make_request ( action = '<STR_LIT>' , <EOL> body = json . dumps ( params ) ) <EOL> def put_configuration_recorder ( self , configuration_recorder ) : <EOL> """<STR_LIT>""" <EOL> params = { '<STR_LIT>' : configuration_recorder , } <EOL> return self . make_request ( action = '<STR_LIT>' , <EOL> body = json . dumps ( params ) ) <EOL> def put_delivery_channel ( self , delivery_channel ) : <EOL> """<STR_LIT>""" <EOL> params = { '<STR_LIT>' : delivery_channel , } <EOL> return self . make_request ( action = '<STR_LIT>' , <EOL> body = json . dumps ( params ) ) <EOL> def start_configuration_recorder ( self , configuration_recorder_name ) : <EOL> """<STR_LIT>""" <EOL> params = { <EOL> '<STR_LIT>' : configuration_recorder_name , <EOL> } <EOL> return self . make_request ( action = '<STR_LIT>' , <EOL> body = json . dumps ( params ) ) <EOL> def stop_configuration_recorder ( self , configuration_recorder_name ) : <EOL> """<STR_LIT>""" <EOL> params = { <EOL> '<STR_LIT>' : configuration_recorder_name , <EOL> } <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>
