if task_instance.next_method:
            timeout_seconds = (
                task_to_execute.execution_timeout - (timezone.utcnow() - task_instance.start_date)
            ).total_seconds()
        else:
            timeout_seconds = task_to_execute.execution_timeout.total_seconds()
        try:
            
            if timeout_seconds <= 0:
                raise AirflowTaskTimeout()
            
            with timeout(timeout_seconds):
                result = execute_callable(context=context, **execute_callable_kwargs)
        except AirflowTaskTimeout:
            task_to_execute.on_kill()
            raise
    else:
        result = execute_callable(context=context, **execute_callable_kwargs)
    with create_session() as session:
        if task_to_execute.do_xcom_push:
            xcom_value = result
        else:
            xcom_value = None
        if xcom_value is not None:  
            task_instance.xcom_push(key=XCOM_RETURN_KEY, value=xcom_value, session=session)
        _record_task_map_for_downstreams(
            task_instance=task_instance, task=task_orig, value=xcom_value, session=session
        )
    return result
def _refresh_from_db(