Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def test_poll(self, cursor):
query_id, _ = cursor.execute("SELECT * FROM one_row")
future = cursor.poll(query_id)
query_execution = future.result()
self.assertIn(query_execution.state, [AthenaQueryExecution.STATE_QUEUED,
AthenaQueryExecution.STATE_RUNNING,
AthenaQueryExecution.STATE_SUCCEEDED,
AthenaQueryExecution.STATE_FAILED,
AthenaQueryExecution.STATE_CANCELLED])
def test_poll(self, cursor):
query_id, _ = cursor.execute("SELECT * FROM one_row")
future = cursor.poll(query_id)
query_execution = future.result()
self.assertIn(query_execution.state, [AthenaQueryExecution.STATE_QUEUED,
AthenaQueryExecution.STATE_RUNNING,
AthenaQueryExecution.STATE_SUCCEEDED,
AthenaQueryExecution.STATE_FAILED,
AthenaQueryExecution.STATE_CANCELLED])
def _poll(self, query_id):
while True:
query_execution = self._get_query_execution(query_id)
if query_execution.state in [AthenaQueryExecution.STATE_SUCCEEDED,
AthenaQueryExecution.STATE_FAILED,
AthenaQueryExecution.STATE_CANCELLED]:
return query_execution
else:
time.sleep(self._poll_interval)