<s> """<STR_LIT>""" <EOL> try : <EOL> import simplejson as json <EOL> except ImportError : <EOL> import json <EOL> from boto . connection import AWSQueryConnection <EOL> from boto . ec2 . cloudwatch . metric import Metric <EOL> from boto . ec2 . cloudwatch . alarm import MetricAlarm , AlarmHistoryItem <EOL> from boto . ec2 . cloudwatch . datapoint import Datapoint <EOL> from boto . regioninfo import RegionInfo <EOL> import boto <EOL> RegionData = { <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' } <EOL> def regions ( ) : <EOL> """<STR_LIT>""" <EOL> regions = [ ] <EOL> for region_name in RegionData : <EOL> region = RegionInfo ( name = region_name , <EOL> endpoint = RegionData [ region_name ] , <EOL> connection_cls = CloudWatchConnection ) <EOL> regions . append ( region ) <EOL> return regions <EOL> def connect_to_region ( region_name , ** kw_params ) : <EOL> """<STR_LIT>""" <EOL> for region in regions ( ) : <EOL> if region . name == region_name : <EOL> return region . connect ( ** kw_params ) <EOL> return None <EOL> class CloudWatchConnection ( AWSQueryConnection ) : <EOL> APIVersion = boto . config . get ( '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' ) <EOL> DefaultRegionName = boto . config . get ( '<STR_LIT>' , '<STR_LIT>' , <EOL> '<STR_LIT>' ) <EOL> DefaultRegionEndpoint = boto . config . get ( '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' ) <EOL> def __init__ ( self , aws_access_key_id = None , aws_secret_access_key = None , <EOL> is_secure = True , port = None , proxy = None , proxy_port = None , <EOL> proxy_user = None , proxy_pass = None , debug = <NUM_LIT:0> , <EOL> https_connection_factory = None , region = None , path = '<STR_LIT:/>' ) : <EOL> """<STR_LIT>""" <EOL> if not region : <EOL> region = RegionInfo ( self , self . DefaultRegionName , <EOL> self . DefaultRegionEndpoint ) <EOL> self . region = region <EOL> AWSQueryConnection . __init__ ( self , aws_access_key_id , <EOL> aws_secret_access_key , <EOL> is_secure , port , proxy , proxy_port , <EOL> proxy_user , proxy_pass , <EOL> self . region . endpoint , debug , <EOL> https_connection_factory , path ) <EOL> def _required_auth_capability ( self ) : <EOL> return [ '<STR_LIT>' ] <EOL> def build_dimension_param ( self , dimension , params ) : <EOL> for dim_name in dimension : <EOL> dim_value = dimension [ dim_name ] <EOL> if isinstance ( dim_value , basestring ) : <EOL> dim_value = [ dim_value ] <EOL> for i , value in enumerate ( dim_value ) : <EOL> params [ '<STR_LIT>' % ( i + <NUM_LIT:1> ) ] = dim_name <EOL> params [ '<STR_LIT>' % ( i + <NUM_LIT:1> ) ] = value <EOL> def build_list_params ( self , params , items , label ) : <EOL> if isinstance ( items , basestring ) : <EOL> items = [ items ] <EOL> for index , item in enumerate ( items ) : <EOL> i = index + <NUM_LIT:1> <EOL> if isinstance ( item , dict ) : <EOL> for k , v in item . iteritems ( ) : <EOL> params [ label % ( i , '<STR_LIT:Name>' ) ] = k <EOL> if v is not None : <EOL> params [ label % ( i , '<STR_LIT>' ) ] = v <EOL> else : <EOL> params [ label % i ] = item <EOL> def build_put_params ( self , params , name , value = None , timestamp = None , <EOL> unit = None , dimensions = None , statistics = None ) : <EOL> args = ( name , value , unit , dimensions , statistics ) <EOL> length = max ( map ( lambda a : len ( a ) if isinstance ( a , list ) else <NUM_LIT:1> , args ) ) <EOL> def aslist ( a ) : <EOL> if isinstance ( a , list ) : <EOL> if len ( a ) != length : <EOL> raise Exception ( '<STR_LIT>' % length ) <EOL> return a <EOL> return [ a ] * length <EOL> for index , ( n , v , u , d , s ) in enumerate ( zip ( * map ( aslist , args ) ) ) : <EOL> metric_data = { '<STR_LIT>' : n } <EOL> if timestamp : <EOL> metric_data [ '<STR_LIT>' ] = timestamp . isoformat ( ) <EOL> if unit : <EOL> metric_data [ '<STR_LIT>' ] = u <EOL> if dimensions : <EOL> self . build_dimension_param ( d , metric_data ) <EOL> if statistics : <EOL> metric_data [ '<STR_LIT>' ] = s [ '<STR_LIT>' ] <EOL> metric_data [ '<STR_LIT>' ] = s [ '<STR_LIT>' ] <EOL> metric_data [ '<STR_LIT>' ] = s [ '<STR_LIT>' ] <EOL> metric_data [ '<STR_LIT>' ] = s [ '<STR_LIT>' ] <EOL> if value != None : <EOL> msg = '<STR_LIT>' <EOL> msg += '<STR_LIT>' <EOL> boto . log . warn ( msg ) <EOL> elif value != None : <EOL> metric_data [ '<STR_LIT>' ] = v <EOL> else : <EOL> raise Exception ( '<STR_LIT>' ) <EOL> for key , value in metric_data . iteritems ( ) : <EOL> params [ '<STR_LIT>' % ( index + <NUM_LIT:1> , key ) ] = value <EOL> def get_metric_statistics ( self , period , start_time , end_time , metric_name , <EOL> namespace , statistics , dimensions = None , <EOL> unit = None ) : <EOL> """<STR_LIT>""" <EOL> params = { '<STR_LIT>' : period , <EOL> '<STR_LIT>' : metric_name , <EOL> '<STR_LIT>' : namespace , <EOL> '<STR_LIT>' : start_time . isoformat ( ) , <EOL> '<STR_LIT>' : end_time . isoformat ( ) } <EOL> self . build_list_params ( params , statistics , '<STR_LIT>' ) <EOL> if dimensions : <EOL> self . build_dimension_param ( dimensions , params ) <EOL> return self . get_list ( '<STR_LIT>' , params , <EOL> [ ( '<STR_LIT>' , Datapoint ) ] ) <EOL> def list_metrics ( self , next_token = None , dimensions = None , <EOL> metric_name = None , namespace = None ) : <EOL> """<STR_LIT>""" <EOL> params = { } <EOL> if next_token : <EOL> params [ '<STR_LIT>' ] = next_token <EOL> if dimensions : <EOL> self . build_dimension_param ( dimensions , params ) <EOL> if metric_name : <EOL> params [ '<STR_LIT>' ] = metric_name <EOL> if namespace : <EOL> params [ '<STR_LIT>' ] = namespace <EOL> return self . get_list ( '<STR_LIT>' , params , [ ( '<STR_LIT>' , Metric ) ] ) <EOL> def put_metric_data ( self , namespace , name , value = None , timestamp = None , <EOL> unit = None , dimensions = None , statistics = None ) : <EOL> """<STR_LIT>""" <EOL> params = { '<STR_LIT>' : namespace } <EOL> self . build_put_params ( params , name , value = value , timestamp = timestamp , <EOL> unit = unit , dimensions = dimensions , statistics = statistics ) <EOL> return self . get_status ( '<STR_LIT>' , params ) <EOL> def describe_alarms ( self , action_prefix = None , alarm_name_prefix = None , <EOL> alarm_names = None , max_records = None , state_value = None , <EOL> next_token = None ) : <EOL> """<STR_LIT>""" <EOL> params = { } <EOL> if action_prefix : <EOL> params [ '<STR_LIT>' ] = action_prefix <EOL> if alarm_name_prefix : <EOL> params [ '<STR_LIT>' ] = alarm_name_prefix <EOL> elif alarm_names : <EOL> self . build_list_params ( params , alarm_names , '<STR_LIT>' ) <EOL> if max_records : <EOL> params [ '<STR_LIT>' ] = max_records <EOL> if next_token : <EOL> params [ '<STR_LIT>' ] = next_token <EOL> if state_value : <EOL> params [ '<STR_LIT>' ] = state_value <EOL> return self . get_list ( '<STR_LIT>' , params , <EOL> [ ( '<STR_LIT>' , MetricAlarm ) ] ) <EOL> def describe_alarm_history ( self , alarm_name = None , <EOL> start_date = None , end_date = None , <EOL> max_records = None , history_item_type = None , <EOL> next_token = None ) : <EOL> """<STR_LIT>""" <EOL> params = { } <EOL> if alarm_name : <EOL> params [ '<STR_LIT>' ] = alarm_name <EOL> if start_date : <EOL> params [ '<STR_LIT>' ] = start_date . isoformat ( ) <EOL> if end_date : <EOL> params [ '<STR_LIT>' ] = end_date . isoformat ( ) <EOL> if history_item_type : <EOL> params [ '<STR_LIT>' ] = history_item_type <EOL> if max_records : <EOL> params [ '<STR_LIT>' ] = max_records <EOL> if next_token : <EOL> params [ '<STR_LIT>' ] = next_token <EOL> return self . get_list ( '<STR_LIT>' , params , <EOL> [ ( '<STR_LIT>' , AlarmHistoryItem ) ] ) <EOL> def describe_alarms_for_metric ( self , metric_name , namespace , period = None , <EOL> statistic = None , dimensions = None , unit = None ) : <EOL> """<STR_LIT>""" <EOL> params = { '<STR_LIT>' : metric_name , <EOL> '<STR_LIT>' : namespace } <EOL> if period : <EOL> params [ '<STR_LIT>' ] = period <EOL> if statistic : <EOL> params [ '<STR_LIT>' ] = statistic <EOL> if dimensions : <EOL> self . build_dimension_param ( dimensions , params ) <EOL> if unit : <EOL> params [ '<STR_LIT>' ] = unit <EOL> return self . get_list ( '<STR_LIT>' , params , <EOL> [ ( '<STR_LIT>' , MetricAlarm ) ] ) <EOL> def put_metric_alarm ( self , alarm ) : <EOL> """<STR_LIT>""" <EOL> params = { <EOL> '<STR_LIT>' : alarm . name , <EOL> '<STR_LIT>' : alarm . metric , <EOL> '<STR_LIT>' : alarm . namespace , <EOL> '<STR_LIT>' : alarm . statistic , <EOL> '<STR_LIT>' : alarm . comparison , <EOL> '<STR_LIT>' : alarm . threshold , <EOL> '<STR_LIT>' : alarm . evaluation_periods , <EOL> '<STR_LIT>' : alarm . period , <EOL> } <EOL> if alarm . actions_enabled is not None : <EOL> params [ '<STR_LIT>' ] = alarm . actions_enabled <EOL> if alarm . alarm_actions : <EOL> self . build_list_params ( params , alarm . alarm_actions , <EOL> '<STR_LIT>' ) <EOL> if alarm . description : <EOL> params [ '<STR_LIT>' ] = alarm . description <EOL> if alarm . dimensions : <EOL> self . build_dimension_param ( alarm . dimensions , params ) <EOL> if alarm . insufficient_data_actions : <EOL> self . build_list_params ( params , alarm . insufficient_data_actions , <EOL> '<STR_LIT>' ) <EOL> if alarm . ok_actions : <EOL> self . build_list_params ( params , alarm . ok_actions , <EOL> '<STR_LIT>' ) <EOL> if alarm . unit : <EOL> params [ '<STR_LIT>' ] = alarm . unit <EOL> alarm . connection = self <EOL> return self . get_status ( '<STR_LIT>' , params ) <EOL> create_alarm = put_metric_alarm <EOL> update_alarm = put_metric_alarm <EOL> def delete_alarms ( self , alarms ) : <EOL> """<STR_LIT>""" <EOL> params = { } <EOL> self . build_list_params ( params , alarms , '<STR_LIT>' ) <EOL> return self . get_status ( '<STR_LIT>' , params ) <EOL> def set_alarm_state ( self , alarm_name , state_reason , state_value , <EOL> state_reason_data = None ) : <EOL> """<STR_LIT>""" <EOL> params = { '<STR_LIT>' : alarm_name , <EOL> '<STR_LIT>' : state_reason , <EOL> '<STR_LIT>' : state_value } <EOL> if state_reason_data : <EOL> params [ '<STR_LIT>' ] = json . dumps ( state_reason_data ) <EOL> return self . get_status ( '<STR_LIT>' , params ) <EOL> def enable_alarm_actions ( self , alarm_names ) : <EOL> """<STR_LIT>""" <EOL> params = { } <EOL> self . build_list_params ( params , alarm_names , '<STR_LIT>' ) <EOL> return self . get_status ( '<STR_LIT>' , params ) <EOL> def disable_alarm_actions ( self , alarm_names ) : <EOL> """<STR_LIT>""" <EOL> params = { } <EOL> self . build_list_params ( params , alarm_names , '<STR_LIT>' ) <EOL> return self . get_status ( '<STR_LIT>' , params ) </s>
