<s> from airflow . utils . decorators import apply_defaults <EOL> from airflow . models import BaseOperator <EOL> from airflow . exceptions import AirflowException <EOL> import logging <EOL> import requests <EOL> import json <EOL> class HipChatAPIOperator ( BaseOperator ) : <EOL> """<STR_LIT>""" <EOL> @ apply_defaults <EOL> def __init__ ( self , <EOL> token , <EOL> base_url = '<STR_LIT>' , <EOL> * args , <EOL> ** kwargs ) : <EOL> super ( HipChatAPIOperator , self ) . __init__ ( * args , ** kwargs ) <EOL> self . token = token <EOL> self . base_url = base_url <EOL> self . method = None <EOL> self . url = None <EOL> self . body = None <EOL> def prepare_request ( self ) : <EOL> """<STR_LIT>""" <EOL> pass <EOL> def execute ( self , context ) : <EOL> self . prepare_request ( ) <EOL> response = requests . request ( self . method , <EOL> self . url , <EOL> headers = { <EOL> '<STR_LIT:Content-Type>' : '<STR_LIT:application/json>' , <EOL> '<STR_LIT>' : '<STR_LIT>' % self . token } , <EOL> data = self . body ) <EOL> if response . status_code >= <NUM_LIT> : <EOL> logging . error ( '<STR_LIT>' , <EOL> response . status_code , response . reason ) <EOL> raise AirflowException ( '<STR_LIT>' % <EOL> ( response . status_code , response . reason ) ) <EOL> class HipChatAPISendRoomNotificationOperator ( HipChatAPIOperator ) : <EOL> """<STR_LIT>""" <EOL> template_fields = ( '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT:message>' ) <EOL> ui_color = '<STR_LIT>' <EOL> @ apply_defaults <EOL> def __init__ ( self , room_id , message , * args , ** kwargs ) : <EOL> super ( HipChatAPISendRoomNotificationOperator , self ) . __init__ ( * args , ** kwargs ) <EOL> self . room_id = room_id <EOL> self . message = message <EOL> default_options = { <EOL> '<STR_LIT>' : '<STR_LIT:html>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : None , <EOL> '<STR_LIT>' : False , <EOL> '<STR_LIT>' : None <EOL> } <EOL> for ( prop , default ) in default_options . iteritems ( ) : <EOL> setattr ( self , prop , kwargs . get ( prop , default ) ) <EOL> def prepare_request ( self ) : <EOL> params = { <EOL> '<STR_LIT:message>' : self . message , <EOL> '<STR_LIT>' : self . message_format , <EOL> '<STR_LIT>' : self . color , <EOL> '<STR_LIT>' : self . frm , <EOL> '<STR_LIT>' : self . attach_to , <EOL> '<STR_LIT>' : self . notify , <EOL> '<STR_LIT>' : self . card <EOL> } <EOL> self . method = '<STR_LIT:POST>' <EOL> self . url = '<STR_LIT>' % ( self . base_url , self . room_id ) <EOL> self . body = json . dumps ( dict ( <EOL> ( k . encode ( '<STR_LIT:utf-8>' ) , v . encode ( '<STR_LIT:utf-8>' ) ) for k , v in params . iteritems ( ) if v ) ) </s>
