how to import boto3 athena client exceptions

Question

I am working with athena from within my python code, using boto3, as follows:

def query_athena(query, output_path):
    client = boto3.client('athena')
    client.start_query_execution(
        ResultConfiguration={'OutputLocation': output_path},
        QueryString=query
    )

As stated in the docs, start_query_execution may raise InternalServerException, InvalidRequestException or TooManyRequestsException. I'd like to treat this as follows:

def query_athena(query, output_path):
    client = boto3.client('athena')
    try:
        client.start_query_execution(
            ResultConfiguration={'OutputLocation': output_path},
            QueryString=query
        )
    except <AthenaException> as e:
        deal with e

where <AthenaException> being one of the three exceptions I mentioned or, better yet, their superclass.

My question is how do I import these exceptions? The docs show them as Athena.Client.exceptions.InternalServerException, but I can't seem to find this Athena.Client in any boto3 module.

Answer 1

I ran into the same confusion, but figured it out. The exceptions listed in the docs aren't internal to boto3, but rather contained in the response when boto3 throws a client error.

My first shot at a solution looks like this. It assumes you've handled s3 output location, a boto3 session, etc already:

import boto3
from botocore.exceptions import ClientError

try:
    client = session.client('athena')
    response = client.start_query_execution(
        QueryString=q,
        QueryExecutionContext={
            'Database': database
        },
        ResultConfiguration={
            'OutputLocation': s3_output,
        }
    )
    filename = response['QueryExecutionId']
    print('Execution ID: ' + response['QueryExecutionId'])

except ClientError as e:
    response = e.response
    code = response['Error']['Code']
    message = response['Error']['Message']
    if code == 'InvalidRequestException':
        print(f'Error in query, {code}:\n{message}')
        raise e
    elif code == 'InternalServerException':
        print(f'AWS {code}:\n{message}')
        raise e
    elif code == 'TooManyRequestsException':
        # Handle a wait, retry, etc here
        pass