<s> import boto <EOL> from boto . connection import AWSQueryConnection <EOL> from boto . regioninfo import RegionInfo <EOL> from boto . exception import JSONResponseError <EOL> from boto . logs import exceptions <EOL> from boto . compat import json <EOL> class CloudWatchLogsConnection ( 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 . LimitExceededException , <EOL> "<STR_LIT>" : exceptions . DataAlreadyAcceptedException , <EOL> "<STR_LIT>" : exceptions . ResourceInUseException , <EOL> "<STR_LIT>" : exceptions . ServiceUnavailableException , <EOL> "<STR_LIT>" : exceptions . InvalidParameterException , <EOL> "<STR_LIT>" : exceptions . ResourceNotFoundException , <EOL> "<STR_LIT>" : exceptions . ResourceAlreadyExistsException , <EOL> "<STR_LIT>" : exceptions . OperationAbortedException , <EOL> "<STR_LIT>" : exceptions . InvalidSequenceTokenException , <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 ( CloudWatchLogsConnection , self ) . __init__ ( ** kwargs ) <EOL> self . region = region <EOL> def _required_auth_capability ( self ) : <EOL> return [ '<STR_LIT>' ] <EOL> def create_log_group ( self , log_group_name ) : <EOL> """<STR_LIT>""" <EOL> params = { '<STR_LIT>' : log_group_name , } <EOL> return self . make_request ( action = '<STR_LIT>' , <EOL> body = json . dumps ( params ) ) <EOL> def create_log_stream ( self , log_group_name , log_stream_name ) : <EOL> """<STR_LIT>""" <EOL> params = { <EOL> '<STR_LIT>' : log_group_name , <EOL> '<STR_LIT>' : log_stream_name , <EOL> } <EOL> return self . make_request ( action = '<STR_LIT>' , <EOL> body = json . dumps ( params ) ) <EOL> def delete_log_group ( self , log_group_name ) : <EOL> """<STR_LIT>""" <EOL> params = { '<STR_LIT>' : log_group_name , } <EOL> return self . make_request ( action = '<STR_LIT>' , <EOL> body = json . dumps ( params ) ) <EOL> def delete_log_stream ( self , log_group_name , log_stream_name ) : <EOL> """<STR_LIT>""" <EOL> params = { <EOL> '<STR_LIT>' : log_group_name , <EOL> '<STR_LIT>' : log_stream_name , <EOL> } <EOL> return self . make_request ( action = '<STR_LIT>' , <EOL> body = json . dumps ( params ) ) <EOL> def delete_metric_filter ( self , log_group_name , filter_name ) : <EOL> """<STR_LIT>""" <EOL> params = { <EOL> '<STR_LIT>' : log_group_name , <EOL> '<STR_LIT>' : filter_name , <EOL> } <EOL> return self . make_request ( action = '<STR_LIT>' , <EOL> body = json . dumps ( params ) ) <EOL> def delete_retention_policy ( self , log_group_name ) : <EOL> """<STR_LIT>""" <EOL> params = { '<STR_LIT>' : log_group_name , } <EOL> return self . make_request ( action = '<STR_LIT>' , <EOL> body = json . dumps ( params ) ) <EOL> def describe_log_groups ( self , log_group_name_prefix = None , <EOL> next_token = None , limit = None ) : <EOL> """<STR_LIT>""" <EOL> params = { } <EOL> if log_group_name_prefix is not None : <EOL> params [ '<STR_LIT>' ] = log_group_name_prefix <EOL> if next_token is not None : <EOL> params [ '<STR_LIT>' ] = next_token <EOL> if limit is not None : <EOL> params [ '<STR_LIT>' ] = limit <EOL> return self . make_request ( action = '<STR_LIT>' , <EOL> body = json . dumps ( params ) ) <EOL> def describe_log_streams ( self , log_group_name , <EOL> log_stream_name_prefix = None , next_token = None , <EOL> limit = None ) : <EOL> """<STR_LIT>""" <EOL> params = { '<STR_LIT>' : log_group_name , } <EOL> if log_stream_name_prefix is not None : <EOL> params [ '<STR_LIT>' ] = log_stream_name_prefix <EOL> if next_token is not None : <EOL> params [ '<STR_LIT>' ] = next_token <EOL> if limit is not None : <EOL> params [ '<STR_LIT>' ] = limit <EOL> return self . make_request ( action = '<STR_LIT>' , <EOL> body = json . dumps ( params ) ) <EOL> def describe_metric_filters ( self , log_group_name , <EOL> filter_name_prefix = None , next_token = None , <EOL> limit = None ) : <EOL> """<STR_LIT>""" <EOL> params = { '<STR_LIT>' : log_group_name , } <EOL> if filter_name_prefix is not None : <EOL> params [ '<STR_LIT>' ] = filter_name_prefix <EOL> if next_token is not None : <EOL> params [ '<STR_LIT>' ] = next_token <EOL> if limit is not None : <EOL> params [ '<STR_LIT>' ] = limit <EOL> return self . make_request ( action = '<STR_LIT>' , <EOL> body = json . dumps ( params ) ) <EOL> def get_log_events ( self , log_group_name , log_stream_name , <EOL> start_time = None , end_time = None , next_token = None , <EOL> limit = None , start_from_head = None ) : <EOL> """<STR_LIT>""" <EOL> params = { <EOL> '<STR_LIT>' : log_group_name , <EOL> '<STR_LIT>' : log_stream_name , <EOL> } <EOL> if start_time is not None : <EOL> params [ '<STR_LIT>' ] = start_time <EOL> if end_time is not None : <EOL> params [ '<STR_LIT>' ] = end_time <EOL> if next_token is not None : <EOL> params [ '<STR_LIT>' ] = next_token <EOL> if limit is not None : <EOL> params [ '<STR_LIT>' ] = limit <EOL> if start_from_head is not None : <EOL> params [ '<STR_LIT>' ] = start_from_head <EOL> return self . make_request ( action = '<STR_LIT>' , <EOL> body = json . dumps ( params ) ) <EOL> def put_log_events ( self , log_group_name , log_stream_name , log_events , <EOL> sequence_token = None ) : <EOL> """<STR_LIT>""" <EOL> params = { <EOL> '<STR_LIT>' : log_group_name , <EOL> '<STR_LIT>' : log_stream_name , <EOL> '<STR_LIT>' : log_events , <EOL> } <EOL> if sequence_token is not None : <EOL> params [ '<STR_LIT>' ] = sequence_token <EOL> return self . make_request ( action = '<STR_LIT>' , <EOL> body = json . dumps ( params ) ) <EOL> def put_metric_filter ( self , log_group_name , filter_name , filter_pattern , <EOL> metric_transformations ) : <EOL> """<STR_LIT>""" <EOL> params = { <EOL> '<STR_LIT>' : log_group_name , <EOL> '<STR_LIT>' : filter_name , <EOL> '<STR_LIT>' : filter_pattern , <EOL> '<STR_LIT>' : metric_transformations , <EOL> } <EOL> return self . make_request ( action = '<STR_LIT>' , <EOL> body = json . dumps ( params ) ) <EOL> def put_retention_policy ( self , log_group_name , retention_in_days ) : <EOL> """<STR_LIT>""" <EOL> params = { <EOL> '<STR_LIT>' : log_group_name , <EOL> '<STR_LIT>' : retention_in_days , <EOL> } <EOL> return self . make_request ( action = '<STR_LIT>' , <EOL> body = json . dumps ( params ) ) <EOL> def set_retention ( self , log_group_name , retention_in_days ) : <EOL> """<STR_LIT>""" <EOL> params = { <EOL> '<STR_LIT>' : log_group_name , <EOL> '<STR_LIT>' : retention_in_days , <EOL> } <EOL> return self . make_request ( action = '<STR_LIT>' , <EOL> body = json . dumps ( params ) ) <EOL> def test_metric_filter ( self , filter_pattern , log_event_messages ) : <EOL> """<STR_LIT>""" <EOL> params = { <EOL> '<STR_LIT>' : filter_pattern , <EOL> '<STR_LIT>' : log_event_messages , <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>
