<s> import os <EOL> import json <EOL> import requests <EOL> import logging <EOL> try : <EOL> from urllib . parse import urlencode <EOL> except ImportError : <EOL> from urllib import urlencode <EOL> from requests . auth import AuthBase <EOL> LOG = logging . getLogger ( __name__ ) <EOL> DEFAULT_TIMEOUT = <NUM_LIT> <EOL> class ApiAuth ( AuthBase ) : <EOL> def __init__ ( self , key ) : <EOL> self . key = key <EOL> def __call__ ( self , r ) : <EOL> r . headers [ '<STR_LIT>' ] = '<STR_LIT>' % self . key <EOL> return r <EOL> class ApiClient ( object ) : <EOL> def __init__ ( self , endpoint = None , key = None ) : <EOL> self . endpoint = endpoint or os . environ . get ( '<STR_LIT>' , "<STR_LIT>" ) <EOL> self . key = key or os . environ . get ( '<STR_LIT>' , '<STR_LIT>' ) <EOL> self . session = requests . Session ( ) <EOL> def __repr__ ( self ) : <EOL> return '<STR_LIT>' % ( self . endpoint , self . key ) <EOL> def get_alerts ( self , query = None ) : <EOL> return self . _get ( '<STR_LIT>' , query ) <EOL> def get_counts ( self , query ) : <EOL> return self . _get ( '<STR_LIT>' , query = None ) <EOL> def get_history ( self , query ) : <EOL> return self . _get ( '<STR_LIT>' , query = None ) <EOL> def send_alert ( self , alert ) : <EOL> return self . _post ( '<STR_LIT>' , data = str ( alert ) ) <EOL> def send ( self , msg ) : <EOL> if msg . event_type == '<STR_LIT>' : <EOL> return self . send_heartbeat ( msg ) <EOL> else : <EOL> return self . send_alert ( msg ) <EOL> def get_alert ( self , alertid ) : <EOL> return self . _get ( '<STR_LIT>' % alertid ) <EOL> def tag_alert ( self , alertid , tags ) : <EOL> if not isinstance ( tags , list ) : <EOL> raise <EOL> return self . _post ( '<STR_LIT>' % alertid , data = json . dumps ( { "<STR_LIT>" : tags } ) ) <EOL> def untag_alert ( self , alertid , tags ) : <EOL> if not isinstance ( tags , list ) : <EOL> raise <EOL> return self . _post ( '<STR_LIT>' % alertid , data = json . dumps ( { "<STR_LIT>" : tags } ) ) <EOL> def open_alert ( self , alertid ) : <EOL> self . update_status ( alertid , '<STR_LIT>' ) <EOL> def ack_alert ( self , alertid ) : <EOL> self . update_status ( alertid , '<STR_LIT>' ) <EOL> def unack_alert ( self , alertid ) : <EOL> self . open_alert ( alertid ) <EOL> def assign_alert ( self , alertid ) : <EOL> self . update_status ( alertid , '<STR_LIT>' ) <EOL> def close_alert ( self , alertid ) : <EOL> self . update_status ( alertid , '<STR_LIT>' ) <EOL> def update_status ( self , alertid , status ) : <EOL> return self . _post ( '<STR_LIT>' % alertid , data = json . dumps ( { "<STR_LIT:status>" : status } ) ) <EOL> def delete_alert ( self , alertid ) : <EOL> return self . _delete ( '<STR_LIT>' % alertid ) <EOL> def send_heartbeat ( self , heartbeat ) : <EOL> """<STR_LIT>""" <EOL> return self . _post ( '<STR_LIT>' , data = str ( heartbeat ) ) <EOL> def get_heartbeats ( self ) : <EOL> """<STR_LIT>""" <EOL> return self . _get ( '<STR_LIT>' ) <EOL> def delete_heartbeat ( self , heartbeatid ) : <EOL> return self . _delete ( '<STR_LIT>' % heartbeatid ) <EOL> def get_top10 ( self , group = '<STR_LIT>' ) : <EOL> return self . _get ( '<STR_LIT>' , { '<STR_LIT>' : group } ) <EOL> def blackout_alerts ( self , blackout ) : <EOL> """<STR_LIT>""" <EOL> return self . _post ( '<STR_LIT>' , data = json . dumps ( blackout ) ) <EOL> def get_blackouts ( self ) : <EOL> """<STR_LIT>""" <EOL> return self . _get ( '<STR_LIT>' ) <EOL> def delete_blackout ( self , blackoutid ) : <EOL> return self . _delete ( '<STR_LIT>' % blackoutid ) <EOL> def get_users ( self , query ) : <EOL> return self . _get ( '<STR_LIT>' , query ) <EOL> def update_user ( self , user , data ) : <EOL> return self . _put ( '<STR_LIT>' % user , data = json . dumps ( data ) ) <EOL> def create_key ( self , key ) : <EOL> return self . _post ( '<STR_LIT>' , data = json . dumps ( key ) ) <EOL> def get_keys ( self ) : <EOL> return self . _get ( '<STR_LIT>' ) <EOL> def revoke_key ( self , key ) : <EOL> return self . _delete ( '<STR_LIT>' % key ) <EOL> def get_status ( self ) : <EOL> return self . _get ( '<STR_LIT>' ) <EOL> def _get ( self , path , query = None ) : <EOL> query = query or tuple ( ) <EOL> url = self . endpoint + path + '<STR_LIT:?>' + urlencode ( query , doseq = True ) <EOL> response = self . session . get ( url , auth = ApiAuth ( self . key ) ) <EOL> LOG . debug ( '<STR_LIT>' , response . headers [ '<STR_LIT>' ] ) <EOL> LOG . debug ( '<STR_LIT>' , response . headers ) <EOL> LOG . debug ( '<STR_LIT>' , response . text ) <EOL> try : <EOL> response . raise_for_status ( ) <EOL> except requests . HTTPError : <EOL> raise <EOL> return response . json ( ) <EOL> def _post ( self , path , data = None ) : <EOL> url = self . endpoint + path <EOL> headers = { '<STR_LIT:Content-Type>' : '<STR_LIT:application/json>' } <EOL> LOG . debug ( '<STR_LIT>' , headers ) <EOL> LOG . debug ( '<STR_LIT>' , data ) <EOL> response = self . session . post ( url , data = data , auth = ApiAuth ( self . key ) , headers = headers ) <EOL> return response . json ( ) <EOL> def _put ( self , path , data = None ) : <EOL> url = self . endpoint + path <EOL> headers = { '<STR_LIT:Content-Type>' : '<STR_LIT:application/json>' } <EOL> LOG . debug ( '<STR_LIT>' , headers ) <EOL> LOG . debug ( '<STR_LIT>' , data ) <EOL> response = self . session . put ( url , data = data , auth = ApiAuth ( self . key ) , headers = headers ) <EOL> return response . json ( ) <EOL> def _delete ( self , path ) : <EOL> url = self . endpoint + path <EOL> response = self . session . delete ( url , auth = ApiAuth ( self . key ) ) <EOL> return response . json ( ) </s>
