Tags: #boto #boto3 #python #cognito #aws
"""
pipenv --python 3.7
pipenv shell
pip install boto3 structlog ipython
"""
import boto3
from botocore.exceptions import ClientError
sdk_client = boto3.client('cognito-idp', **{'region_name': 'eu-west-1'})
import logging
import structlog
structlogger = structlog.getLogger(__name__)
structlog.configure(cache_logger_on_first_use=True,
context_class=dict,
logger_factory=structlog.stdlib.LoggerFactory(),
processors=[structlog.stdlib.add_logger_name,
structlog.stdlib.add_log_level,
structlog.processors.TimeStamper(fmt="%Y-%m-%d %H:%M.%S"),
structlog.processors.JSONRenderer(sort_keys=True)])
try:
sdk_client.global_sign_out(AccessToken="123")
except TypeError as err:
structlogger.error('TypeError Caught',
exc_type=err.__class__.__name__,
exc_msg=str(err),
orig_err=err)
logging.error("Stdlib Logger: TypeError Caught: %s", err, exc_info=True)
except ClientError as err:
structlogger.error('ClientError Caught',
exc_type=err.__class__.__name__,
exc_msg=str(err),
orig_err=err)
logging.error("Stdlib Logger: ClientError Caught: %s", err, exc_info=True)
except Exception as err:
structlogger.error('Exception Caught',
exc_type=err.__class__.__name__,
exc_msg=str(err),
orig_err=err)
logging.error("Stdlib Logger: Exception Caught: %s", err, exc_info=False) # hide stack trace
###################################################################################################
aws_access_key = '111'
aws_secret_key = '222'
region = 'us-east-1'
try:
iam = boto3.client('iam', **{'aws_access_key_id': aws_access_key,
'aws_secret_access_key': aws_secret_key,
'region_name': region})
user = iam.create_user(UserName='foobar')
print(f"created user! {user}")
except TypeError as err:
structlogger.error('TypeError Caught',
exc_class=type(err),
exc_type=err.__class__.__name__,
exc_msg=str(err),
err_code=err.response['ResponseMetadata']['HTTPStatusCode'] if hasattr(err, 'response') else 500)
logging.error("Stdlib Logger: TypeError Caught: %s", err, exc_info=True)
except ClientError as err:
structlogger.error('ClientError Caught',
exc_class=type(err),
exc_type=err.__class__.__name__,
exc_msg=str(err),
err_code=err.response['ResponseMetadata']['HTTPStatusCode'] if hasattr(err, 'response') else 500)
logging.error("Stdlib Logger: ClientError Caught: %s", err, exc_info=True)
except Exception as err:
structlogger.error('Exception Caught',
exc_class=type(err),
exc_type=err.__class__.__name__,
exc_msg=str(err),
err_code=err.response['ResponseMetadata']['HTTPStatusCode'] if hasattr(err, 'response') else 500)
logging.error("Stdlib Logger: Exception Caught: %s", err, exc_info=False) # hide stack trace
"""
log output...
{
"err_code": 403,
"event": "Exception Caught",
"exc_msg": "An error occurred (AccessDenied) when calling the CreateUser operation: User: arn:aws:iam::123:user/beep is not authorized to perform: iam:CreateUser on resource: arn:aws:iam::123:user/foobar",
"exc_class": "<class 'botocore.exceptions.ClientError'>",
"exc_type": "ClientError",
"level": "error",
"logger": "__main__",
"timestamp": "2018-08-31 11:46.23"
}
err.response...
{
"err_code":{
"Error":{
"Code":"AccessDenied",
"Message":"User: arn:aws:iam::123:user/beep is not authorized to perform: iam:CreateUser on resource: arn:aws:iam::123:user/foobar",
"Type":"Sender"
},
"ResponseMetadata":{
"HTTPHeaders":{
"content-length":"405",
"content-type":"text/xml",
"date":"Fri, 31 Aug 2018 11:30:15 GMT",
"x-amzn-requestid":"456"
},
"HTTPStatusCode":403,
"RequestId":"456",
"RetryAttempts":0
}
},
"err_type":"<class 'botocore.exceptions.ClientError'>",
"event":"ClientError Caught",
"exc_msg":"An error occurred (AccessDenied) when calling the CreateUser operation: User: arn:aws:iam::123:user/beep is not authorized to perform: iam:CreateUser on resource: arn:aws:iam::123:user/foobar",
"exc_type":"ClientError",
"level":"error",
"logger":"__main__",
"orig_err":"ClientError('An error occurred (AccessDenied) when calling the CreateUser operation: User: arn:aws:iam::123:user/beep is not authorized to perform: iam:CreateUser on resource: arn:aws:iam::123:user/foobar')",
"timestamp":"2018-08-31 11:30.16"
}
See also...
try:
# some aws api call
except ClientError as exc:
error = exc.response.get('Error', {})
error_code = error.get('Code')
error_msg = error.get('Message')
invalid_refresh_token = re.compile('Invalid Refresh Token')
if error_code == 'NotAuthorizedException' and invalid_refresh_token.match(error_msg):
instr_exc(exc, msg, state='expired', **exc_tags)
raise exceptions.CognitoException(msg, code=extract_status_code(exc))
instr_exc(exc, msg, state='failed', **exc_tags)
raise exceptions.CognitoException(msg, code=extract_status_code(exc))
...and...
def extract_status_code(exc):
"""If dealing with an AWS SDK response, attempt to extract the status code."""
return exc.response['ResponseMetadata']['HTTPStatusCode'] if hasattr(exc, 'response') else 500
"""