Boto3, python and error handling methods

Question

I just picked up python as my scripting language, and I'm trying to figure out how to handle errors correctly using boto3.

I am trying to create an IAM user:

def create_user(username, iam_conn): try: user = iam_conn.create_user(UserName=username) return user except Exception as e: return e 

When the create_user call succeeds, I get a neat object that contains the HTTP status code of the API call and the data of the newly created user.

Example:

 {'ResponseMetadata': {'HTTPStatusCode': 200, 'RequestId': 'omitted' }, u'User': {u'Arn': 'arn:aws:iam::omitted:user/omitted', u'CreateDate': datetime.datetime(2015, 10, 11, 17, 13, 5, 882000, tzinfo=tzutc()), u'Path': '/', u'UserId': 'omitted', u'UserName': 'omitted' } } 

This works great. But when this fails (for example, if the user already exists), I just get an object like botocore.exceptions.ClientError with just text to tell me what went wrong.

Example: ClientError ("An error occurred (EntityAlreadyExists) when calling the CreateUser operation: a user with a missing name already exists. ',)

This (AFAIK) makes error handling very difficult because I cannot just include the resulting http status code (409 for the user already exists according to AWS API docs for IAM). It makes me think that I should do something wrong. The best way would be to boto3 never throw exceptions, but juts always return an object that reflects the way the API is called.

Can someone enlighten me on this issue or point me in the right direction?

Thank you so much!

Answer 1

Use the answer contained in the exception. Here is an example:

 import boto3 from botocore.exceptions import ClientError try: iam = boto3.client('iam') user = iam.create_user(UserName='fred') print("Created user: %s" % user) except ClientError as e: if e.response['Error']['Code'] == 'EntityAlreadyExists': print("User already exists") else: print("Unexpected error: %s" % e) 

The dictated response in the exception will contain the following:

• ['Error']['Code'] e.g. 'EntityAlreadyExists' or 'ValidationError'
• ['ResponseMetadata']['HTTPStatusCode'] eg 400
• ['ResponseMetadata']['RequestId'] e.g. 'd2b06652-88d7-11e5-99d0-812348583a35'
• ['Error']['Message'] for example, "An error has occurred (EntityAlreadyExists) ..."
• ['Error']['Type'] for example, 'Sender'

See Botocore Error Handling for more information.

[Updated: 2018-03-07]

The AWS Python SDK began to provide service exceptions on clients (but not on resources ) that you can explicitly catch, so now you can write this code something like this:

 import boto3 from botocore.exceptions import ClientError, ParamValidationError try: iam = boto3.client('iam') user = iam.create_user(UserName='fred') print("Created user: %s" % user) except iam.exceptions.EntityAlreadyExistsException: print("User already exists") except ParamValidationError as e: print("Parameter validation error: %s" % e) except ClientError as e: print("Unexpected error: %s" % e) 

Unfortunately, there is currently no documentation for these exceptions.

Answer 2

Just updating the "no exception on resources" problem that @jarmod points to (please feel free to update your answer if below seems appropriate)

I checked the code below and it works fine. It uses "resources" to execute client.exceptions , but catches client.exceptions - although it "looks" somewhat wrong ... it tests well, exception classes show and map when considering using debugger during an exception ...

This may not apply to all resources and clients, but it works for data folders (the so-called s3 segments).

 lab_session = boto3.Session() c = lab_session.client('s3') #this client is only for exception catching try: b = s3.Bucket(bucket) b.delete() except c.exceptions.NoSuchBucket as e: #ignoring no such bucket exceptions logger.debug("Failed deleting bucket. Continuing. {}".format(e)) except Exception as e: #logging all the others as warning logger.warning("Failed deleting bucket. Continuing. {}".format(e)) 

Hope this helps ...

Answer 3

I found this very useful since the exceptions are not documented to list all the exceptions on the screen for this package. Here is the code I used for this:

 import botocore.exceptions def listexns(mod): #module = __import__(mod) exns = [] for name in botocore.exceptions.__dict__: if (isinstance(botocore.exceptions.__dict__[name], Exception) or name.endswith('Error')): exns.append(name) for name in exns: print('%s.%s is an exception type' % (str(mod), name)) return if __name__ == '__main__': import sys if len(sys.argv) <= 1: print('Give me a module name on the $PYTHONPATH!') print('Looking for exception types in module: %s' % sys.argv[1]) listexns(sys.argv[1]) 

That leads to:

 Looking for exception types in module: boto3 boto3.BotoCoreError is an exception type boto3.DataNotFoundError is an exception type boto3.UnknownServiceError is an exception type boto3.ApiVersionNotFoundError is an exception type boto3.HTTPClientError is an exception type boto3.ConnectionError is an exception type boto3.EndpointConnectionError is an exception type boto3.SSLError is an exception type boto3.ConnectionClosedError is an exception type boto3.ReadTimeoutError is an exception type boto3.ConnectTimeoutError is an exception type boto3.ProxyConnectionError is an exception type boto3.NoCredentialsError is an exception type boto3.PartialCredentialsError is an exception type boto3.CredentialRetrievalError is an exception type boto3.UnknownSignatureVersionError is an exception type boto3.ServiceNotInRegionError is an exception type boto3.BaseEndpointResolverError is an exception type boto3.NoRegionError is an exception type boto3.UnknownEndpointError is an exception type boto3.ConfigParseError is an exception type boto3.MissingParametersError is an exception type boto3.ValidationError is an exception type boto3.ParamValidationError is an exception type boto3.UnknownKeyError is an exception type boto3.RangeError is an exception type boto3.UnknownParameterError is an exception type boto3.AliasConflictParameterError is an exception type boto3.PaginationError is an exception type boto3.OperationNotPageableError is an exception type boto3.ChecksumError is an exception type boto3.UnseekableStreamError is an exception type boto3.WaiterError is an exception type boto3.IncompleteReadError is an exception type boto3.InvalidExpressionError is an exception type boto3.UnknownCredentialError is an exception type boto3.WaiterConfigError is an exception type boto3.UnknownClientMethodError is an exception type boto3.UnsupportedSignatureVersionError is an exception type boto3.ClientError is an exception type boto3.EventStreamError is an exception type boto3.InvalidDNSNameError is an exception type boto3.InvalidS3AddressingStyleError is an exception type boto3.InvalidRetryConfigurationError is an exception type boto3.InvalidMaxRetryAttemptsError is an exception type boto3.StubResponseError is an exception type boto3.StubAssertionError is an exception type boto3.UnStubbedResponseError is an exception type boto3.InvalidConfigError is an exception type boto3.InfiniteLoopConfigError is an exception type boto3.RefreshWithMFAUnsupportedError is an exception type boto3.MD5UnavailableError is an exception type boto3.MetadataRetrievalError is an exception type boto3.UndefinedModelAttributeError is an exception type boto3.MissingServiceIdError is an exception type 

Answer 4

Or a comparison by class name, for example

 except ClientError as e: if 'EntityAlreadyExistsException' == e.__class__.__name__: # handle specific error 

Since they are dynamically created, you can never import a class and catch it using real Python.

Answer 5

If you call the sign_up API (AWS Cognito) using Python3, you can use the following code.

 def registerUser(userObj): ''' Registers the user to AWS Cognito. ''' # Mobile number is not a mandatory field. if(len(userObj['user_mob_no']) == 0): mobilenumber = '' else: mobilenumber = userObj['user_country_code']+userObj['user_mob_no'] secretKey = bytes(settings.SOCIAL_AUTH_COGNITO_SECRET, 'latin-1') clientId = settings.SOCIAL_AUTH_COGNITO_KEY digest = hmac.new(secretKey, msg=(userObj['user_name'] + clientId).encode('utf-8'), digestmod=hashlib.sha256 ).digest() signature = base64.b64encode(digest).decode() client = boto3.client('cognito-idp', region_name='eu-west-1' ) try: response = client.sign_up( ClientId=clientId, Username=userObj['user_name'], Password=userObj['password1'], SecretHash=signature, UserAttributes=[ { 'Name': 'given_name', 'Value': userObj['given_name'] }, { 'Name': 'family_name', 'Value': userObj['family_name'] }, { 'Name': 'email', 'Value': userObj['user_email'] }, { 'Name': 'phone_number', 'Value': mobilenumber } ], ValidationData=[ { 'Name': 'email', 'Value': userObj['user_email'] }, ] , AnalyticsMetadata={ 'AnalyticsEndpointId': 'string' }, UserContextData={ 'EncodedData': 'string' } ) except ClientError as error: return {"errorcode": error.response['Error']['Code'] } except Exception as e: return {"errorcode": "Something went wrong. Try later or contact the admin" } return {"success": "User registered successfully. "} 

error.response ['Error'] ['Code'] will be InvalidPasswordException, UsernameExistsException, etc. Thus, in the main function or in the place where you call the function, you can write logic to provide a meaningful message to the user.

Example for the answer (error.response):

 { "Error": { "Message": "Password did not conform with policy: Password must have symbol characters", "Code": "InvalidPasswordException" }, "ResponseMetadata": { "RequestId": "c8a591d5-8c51-4af9-8fad-b38b270c3ca2", "HTTPStatusCode": 400, "HTTPHeaders": { "date": "Wed, 17 Jul 2019 09:38:32 GMT", "content-type": "application/x-amz-json-1.1", "content-length": "124", "connection": "keep-alive", "x-amzn-requestid": "c8a591d5-8c51-4af9-8fad-b38b270c3ca2", "x-amzn-errortype": "InvalidPasswordException:", "x-amzn-errormessage": "Password did not conform with policy: Password must have symbol characters" }, "RetryAttempts": 0 } } 

For further reference: https://boto3.amazonaws.com/v1/documentation/api/latest/reference/services/cognito-idp.html#CognitoIdentityProvider.Client.sign_up

Answer 6

You need to do something when it does not cope with the problem. You are now returning the actual exception. For example, if this is not a problem, that the user already exists, and you want to use it as the get_or_create function, perhaps you can solve the problem by returning the existing user object.

 try: user = iam_conn.create_user(UserName=username) return user except botocore.exceptions.ClientError as e: #this exception could actually be other things other than exists, so you want to evaluate it further in your real code. if e.message.startswith( 'enough of the exception message to identify it as the one you want') print('that user already exists.') user = iam_conn.get_user(UserName=username) return user elif e.message.some_other_condition: #something else else: #unhandled ClientError raise(e) except SomeOtherExceptionTypeYouCareAbout as e: #handle it # any unhandled exception will raise here at this point. # if you want a general handler except Exception as e: #handle it. 

However, this may be a problem for your application, in which case you want to place an exception handler around the code that called your user-defined function, and let the calling function determine how to deal with it, for example, asking the user to enter a different name user or something meaningful to your application.