<s> import logging <EOL> from airflow . exceptions import AirflowException <EOL> from airflow . hooks import HttpHook <EOL> from airflow . models import BaseOperator <EOL> from airflow . utils . decorators import apply_defaults <EOL> class SimpleHttpOperator ( BaseOperator ) : <EOL> """<STR_LIT>""" <EOL> template_fields = ( '<STR_LIT>' , '<STR_LIT:data>' , ) <EOL> template_ext = ( ) <EOL> ui_color = '<STR_LIT>' <EOL> @ apply_defaults <EOL> def __init__ ( self , <EOL> endpoint , <EOL> method = '<STR_LIT:POST>' , <EOL> data = None , <EOL> headers = None , <EOL> response_check = None , <EOL> extra_options = None , <EOL> http_conn_id = '<STR_LIT>' , * args , ** kwargs ) : <EOL> super ( SimpleHttpOperator , self ) . __init__ ( * args , ** kwargs ) <EOL> self . http_conn_id = http_conn_id <EOL> self . method = method <EOL> self . endpoint = endpoint <EOL> self . headers = headers or { } <EOL> self . data = data or { } <EOL> self . response_check = response_check <EOL> self . extra_options = extra_options or { } <EOL> def execute ( self , context ) : <EOL> http = HttpHook ( self . method , http_conn_id = self . http_conn_id ) <EOL> logging . info ( "<STR_LIT>" ) <EOL> response = http . run ( self . endpoint , <EOL> self . data , <EOL> self . headers , <EOL> self . extra_options ) <EOL> if self . response_check : <EOL> if not self . response_check ( response ) : <EOL> raise AirflowException ( "<STR_LIT>" ) </s>
