def check_status(self, job_name, key,<EOL>describe_function, check_interval,<EOL>max_ingestion_time,<EOL>non_terminal_states=None):
if not non_terminal_states:<EOL><INDENT>non_terminal_states = self.non_terminal_states<EOL><DEDENT>sec = <NUM_LIT:0><EOL>running = True<EOL>while running:<EOL><INDENT>time.sleep(check_interval)<EOL>sec = sec + check_interval<EOL>try:<EOL><INDENT>response = describe_function(job_name)<EOL>status = response[key]<EOL>self.log.info('<STR_LIT>'<EOL>'<STR_LIT>' % (sec, status))<EOL><DEDENT>except KeyError:<EOL><INDENT>raise AirflowException('<STR_LIT>')<EOL><DEDENT>except ClientError:<EOL><INDENT>raise AirflowException('<STR_LIT>')<EOL><DEDENT>if status in non_terminal_states:<EOL><INDENT>running = True<EOL><DEDENT>elif status in self.failed_states:<EOL><INDENT>raise AirflowException('<STR_LIT>' % response['<STR_LIT>'])<EOL><DEDENT>else:<EOL><INDENT>running = False<EOL><DEDENT>if max_ingestion_time and sec > max_ingestion_time:<EOL><INDENT>raise AirflowException('<STR_LIT>', max_ingestion_time)<EOL><DEDENT><DEDENT>self.log.info('<STR_LIT>')<EOL>response = describe_function(job_name)<EOL>return response<EOL>
Check status of a SageMaker job

:param job_name: name of the job to check status
:type job_name: str
:param key: the key of the response dict
    that points to the state
:type key: str
:param describe_function: the function used to retrieve the status
:type describe_function: python callable
:param args: the arguments for the function
:param check_interval: the time interval in seconds which the operator
    will check the status of any SageMaker job
:type check_interval: int
:param max_ingestion_time: the maximum ingestion time in seconds. Any
    SageMaker jobs that run longer than this will fail. Setting this to
    None implies no timeout for any SageMaker job.
:type max_ingestion_time: int
:param non_terminal_states: the set of nonterminal states
:type non_terminal_states: set
:return: response of describe call after job is done