AWS: Cloudshell Python version mismatch

 Lately, I tried to install a Python library onto Cloudshell. I first checked the Python version:

$ python --version

Python 3.13.13

So I ran pip install, which deployed a version of my lib compatible with Python 3.13. The lib declares some entry points for setuptools, so that I can run some commands from the CLI. So I run those commands from the shell, but they fail with some errors about missing import.

After investigation, I found out that the command scripts started with a shebang like this one:

#!/usr/bin/python3

This is all normal, except that if I check the link, this is what I find:

$ ls -l /usr/bin/python3

lrwxrwxrwx. 1 root root 9 Apr 20 22:07 /usr/bin/python3-> python3.9

Someone forgot something? Or am I looking at it wrong?

AWS: The State of Account State

 In September 2025, AWS announced that the Account information in the Organizations Service will have a new State field to replace the Status field. Since that date, both fields are available for all Organizations operations, but the Status field is vowed to be removed on September 2026.

When you read such an announcement and you know your code is using the Status field, you project to review your code and update it. So we did quite immediately, but we could not see the new State field when executing our lambdas. So we postponed the update for later.

Recently, I had another look at the problem, and still could not see any State field appearing in lambdas. I tested some call to DescribeAccount within CloudShell, but the field was really there. So I decided to run the following lambda:

import boto3

import botocore

def lambda_handler(event, context):

    print("boto3:", boto3.__version__)

    print("botocore:", botocore.__version__)

    org_client = boto3.client("organizations")

    response = org_client.describe_account(AccountId="123456789012")

    print(response)

I was surprised by the result.

boto3: 1.40.4
botocore: 1.40.4

Those versions were released in August 2025, before the update. CloudShell in my test uses botocore 1.42.72, which is from March this year. When I notified AWS Support about it, they just told me to use a Layer with a more recent botocore included. How long should I keep this temporary workaround?

AWS: Duplicates in Search Provisioned Products

 Using our beloved boto3 library, we are looking for the list of all our Provisioned Products in Service Catalog.

sc_client = boto3.client('servicecatalog')

result = sc_client.search_provisioned_products(

    PageSize=20

)

I won't bore you with th code that loops over the result and perform the operation again if we have more than 20 products. But the strange thing, is that wherever we had more than the page size, some products were repeated in the other pages. Thinking of a bug in AWS Service Catalog, we reached out to the Support Team. This is their answer:

This is a known behavior with the SearchProvisionedProducts API when using the default relevance-based sorting. Because results are sorted by relevance, the ordering can shift slightly between paginated requests, which causes duplicates (or occasionally missed items) across pages.

Never heard of relevance-based sorting. Looking at the documentation, there is no mention of it:

SortBy

The sort field. If no value is specified, the results are not sorted. The valid values are arn, id, name, and lastRecordId.

 Then, Support Team is proposing a solution:

Adding SortBy='createdTime' gives the pagination a stable ordering, so the page token points to a consistent boundary between pages. No more duplicates should appear regardless of how many provisioned products you have.

It is interesting to note that 'createdTime' is not listed in the documentation either. We tried it and it works. So a hidden feature solves a known bug.

AWS: Step functions can keep only Lambda payloads

 When running an AWS Lambda in synchronous mode from a Step Function, you will get the Lambda's return value in the output's Payload part. But unfortunately, you might also get some mostly useless information as well:

{

    "ExecutedVersion": "$LATEST",

    "Payload": {

        "result": "value"

    },

    "SdkHttpMetadata": {},

    "HttpHeaders": {}

}

When browsing your Step Functions's output, that's a lot of noise. So to keep only the lambda's payload part, you can add this line in your task definition:

    "ResultSelector": { "Payload.$": "$.Payload" }

Keep only the good stuff!

AWS: Physical Resource ID in Custom Resources

Originally, Custom Resources in Cloudformation were designed for wrapping AWS resources that are not yet supported by the Cloudformation service into a Lambda. However, we often use them for other purposes:

• Retrieve information from other resources (like the data in terraform)
• Trigger some actions
• Implement some logic

In most of those cases, we do not care about resource deletion. But we are often surprised by calls from Cloudformation to delete the resource. The reason is the misunderstanding or the misuse of the Physical Resource ID. And the origin of this, for me, comes from a bad design choice on AWS part.

Let's have a look at the way the cfnresponse module is written. 

def send(event, context, responseStatus, responseData, physicalResourceId=None, noEcho=False, reason=None):
    responseUrl = event['ResponseURL']
    responseBody = {
        'Status' : responseStatus,
        'Reason' : reason or "See the details in CloudWatch Log Stream: {}".format(context.log_stream_name),
        'PhysicalResourceId' : physicalResourceId or context.log_stream_name,
        'StackId' : event['StackId'],
        'RequestId' : event['RequestId'],
        'LogicalResourceId' : event['LogicalResourceId'],
        'NoEcho' : noEcho,
        'Data' : responseData
    }

There are 2 bad choices:

•  The Physical Resource ID parameter is optional. It makes you think that it is not important. That if you don't set it, some default behavior will handle it correctly for you. 
• The default value is random. Even worse, it is not consistently random. It is set to the log stream name, that changes on each Lambda cold start.

That means that most of the time, your Physical Resource ID will change on each call, except if you trigger it several times in a row. And this change of Physical Resource ID is the one that triggers the call to the delete part of your Lambda.

You can imagine that your Resource behaves like an EC2. If you modify a tag, your instance will keep its ID. But if you change its VPC, a new instance will be created, with a new ID. In that case, the old instance must be deleted. You can consider the Physical Resource ID to be like the instance ID. You want to decide, based on which parameter was modified, if the old Resource must be kept or deleted. 

Which means that in most cases, you do not want your Physical Resource ID to change. So the default behavior is wrong. It will lead to:

• Have your Lambda called for deletion for no reason.
• Can cause accidental calls that you don't expect. We had the case when an S3 bucket was deleted in production because someone added a parameter to the Custom Resource.
• Makes you write some useless code to avoid to call the Lambda when you don't expect it, like checking that your Cloudformation stack is really deleting the Resource. 

So the correct behavior is to always set the Physical Resource ID. And usually to a constant value:

cfnresponse.send(event, context, cfnresponse.SUCCESS, responseData, "ConstantPhysicalID")

What about the legacy code? Those old Custom Resources that already have a Physical Resource ID set to the log stream name? The good thing is that the previously set Physical Resource ID is sent to the Lambda in the event parameter. So you can simply set it back to its previous value:

physicalId = event["PhysicalResourceId"]

cfnresponse.send(event, context, cfnresponse.SUCCESS, responseData, physicalId)

AWS: Why Serverless Macro for Cloudformation always packages?

I'm using the Serverless macro quite a lot in my Cloudformation templates. It is very practical that you point a Lambda content to a local folder. Then Cloudformation packages the whole content into a Zip file and uploads it to an S3 Bucket.

However, it happens that I use the Serverless macro for some other feature, like generating the Event Rule that trigger my Lambda for instance. In some cases, my Lambda code can be already packaged in a Container on ECR, or even inlined. In those cases, I don't need any packaging.

What I noticed, is that Cloudformation is still packaging something. I downloaded the packaged Zip and checked its content. I could find the complete folder from the Cloudformation template location. For one template that was stored in the root of my source code, it packaged the complete application!

Does someone know why is that? Is there a reason for packaging when a Lambda is only inlined? Is there a way to tell Cloudformation to avoid packaging? Is it a bug?

AWS: Find Root Cause of Failure for CloudFormation Stacks

 When a CloudFormation stack fails, you have to scroll back trough the events to find the root cause of the failure. Recently, AWS even added a "Detect Root Cause" button to the Console to immediately scroll to the correct event. But how do you do it from a python script?

import boto3

def find_root_cause(stack_name):

    cf_client = boto3.client('cloudformation')

    next_values = "First Time"

    params = {

        "StackName": stack_name

    }

    root_cause = None

    while next_values:

        result = cf_client.describe_stack_events(**params)

        next_values = result.get("NextToken")

        params["NextToken"] = next_values

        for event in result["StackEvents"]:

            status = event.get("ResourceStatus", "")

            reason = event.get("ResourceStatusReason")

            # start of deployment

            if reason == "User Initiated":

                return root_cause

           

            if reason and "FAILED" in status:

                root_cause = reason

    return root_cause

You follow the same pattern as from the Console. You go back the events history, until you reach the oldest error message before the start of the deployment.

Moto: Alias Issue when Creating S3 Access Point is Fixed

 Two days ago, I discovered a small bug in the moto library that I use to unit test my lambdas on AWS. I needed to create an S3 Access Point with boto3, and while retrieving the its alias, I had different results when using the return value of create_access_point or get_access_point.

So I wrote a small unit test:

from moto import mock_s3control

import boto3

@mock_s3control

def test_access_point_alias():

    client = boto3.client("s3control")

    alias_from_create = client.create_access_point(

        AccountId="123456789012",

        Name="my-access-point",

        Bucket="MyBucket",

    )["Alias"]

    alias_from_get = client.get_access_point(

        AccountId="123456789012",

        Name="my-access-point",

    )["Alias"]

    assert alias_from_create == alias_from_get

I create an S3 Access Point, and retrieve its alias in two ways: from the response of the create_access_point function, and from the get_access_point function. On my moto 4.2.6, this test fails.

So I opened an issue on the project's repository. It was fixed and closed on the same day. That's reactivity!

AWS: The Next Token Pattern

When developing  for AWS, there is a pattern that you use each time a response to a service may return a lot of data. You get back some of the data, together with a token that you can provide to get another round of data.

As an example, let's take the service that returns the list of events from a Cloudformation template deployment. Here is how you would do it using Python and boto3:

cf_client = boto3.client("cloudformation")

response = cf_client.describe_stack_events(StackName="mystack")

# do something with the response

while response.get("NextToken"):

    response = cf_client.describe_stack_events(

        StackName="mystack",

        NextToken=response.get("NextToken")

    )

    # do again something with the response

However, there is one thing that I do not like with this pattern: code repeat. You call the service at two different parts of your code, with almost identical parameters. And you process the response in the same way, again in two places. If you have to fix something in this code, you have to remember to fix in both places.

The approach I use to have your code only once, is to take advantage of Python's capacity to pass parameters as a dictionary. Here is my approach to this pattern:

cf_client = boto3.client("cloudformation")

next_token = "FIRST TIME"

params = {"StackName": "mystack"}

while next_token:

    response = cf_client.describe_stack_events(**params)

    # do something with the response

    next_token = response.get("NextToken")

    params["NextToken"] = next_token

I store my parameter list in a dictionary, and I initialize the next token with something that is not empty. So the first time in the loop will always run. I can then call my service, without the token. After processing the response, I then read the next token and fill it in my parameter list. The second time round, it will call the service with my token.

Something funny happened when I started using this pattern into our production code. We tried using Bandit, which is a tool that analyze your code and looks for security issues. It would systematically flag my pattern with this error: 

[B105:hardcoded_password_string] Possible hardcoded password: 'FIRST TIME'

 Well, I have to slightly modify my pattern to avoid using the word token...

AWS: Simpler S3 File Deletes by Prefix

 I came across this code that deletes files in an S3 from a list of prefixes:

s3_client = boto3.client('s3')

for prefix in prefix_list:

    paginator = s3_client.get_paginator('list_objects_v2')

    file_list = paginator.paginate(

        Bucket=data_bucket,

        Prefix=prefix

    )

    for current_content in file_list:

        for current_file in current_content.get('Contents', []):

            current_key = current_file['Key']

            response = s3_client.delete_object(

                Bucket=data_bucket,

                Key=current_key

            )

The code creates an S3 client, and then, for each prefix in a list, it creates a paginator. Paginators are great because they help you avoid using all your memory when the list of files is big. Using this paginator, the code retrieves the list of all the files corresponding to the prefix, and deletes it.

Nothing bad in the code, it works nicely. My only remark here, is that there exists a simpler way. Instead of using the S3 client, you can create an S3 bucket resource. From there, you can simply delete all files listed under a prefix using a simple filter:

s3_resource = boto3.resource('s3')

bucket = s3_resource.Bucket(data_bucket)

for prefix in prefix_list:

    bucket.objects.filter(Prefix=prefix).delete()

Simpler!

AWS: Automatic Subscription Confirmation from SQS Queue to SNS Topic

 We have an architecture in AWS where different events from different accounts need to be sent to one central SQS queue. Since the events will cross both accounts and regions, one way to do it is to send them to a local SNS Topic. 

The SQS queue will have to subscribe to all those Topics, but we can not do it on the SQS side, since it does not know each time someone pops out a new account. However, the problem with having the SNS Topics create the subscriptions, is that they are waiting for confirmation from the SQS queue.

Since we already have a lambda waiting on the other side of the queue, handling all the events, we added a small code to handle the subscription confirmation as well. Here it is:

import json

import urllib.request

def lambda_handler(event, context):

    for record in event["Records"]:

        body = json.loads(record["body"])

        if body.get("Type") == "SubscriptionConfirmation":

            handle_subscription_confirmation(body)

def handle_subscription_confirmation(message):

    url = message["SubscribeURL"]

    with urllib.request.urlopen(url) as response:

        print(response.read())

I find it strange that the Cloudformation template that we use to create the subscription does not handle the confirmation as well. Or maybe not cross-account?

AWS: Create a Layer with Git

 If you want to perform Git operations inside an AWS Lambda written in Python, you can decide to use the GitPython library. However, it needs git to be installed on your system.

In order to package git, I discovered a really nice tool: lambci/yumda. It's a docker container that overrides the yum installer to deploy the library you install, including all its dependencies, into a separate folder called /lambda/opt. So all you need to do is to yum your git, and zip the content of the folder.

Here is an example of a script that packages both GitPython and git into a zip file compatible with AWS Layers:

docker run --rm -v "$PWD":/tmp/layer lambci/yumda:2 bash -c "

  yum install -y git && \

  cd /lambda/opt && \

  zip -yr /tmp/layer/gitlayer.zip

"

pip install GitPython -t python

zip -r gitlayer.zip python/*

Then you can send this zip to an S3 bucket and create your layer easily. The nice thing about running this script inside a CodeBuild is that all tools, such as docker and pip, are already installed. However, for docker, do not forget to enable the Privileged Mode.

Warnings on SQS endpoint URL while mocking with moto

 We have many unit tests using moto library to mock calls to boto3 to access AWS services.They all work fine, except for a warning on the use of this fixture:

@fixture(scope='function')

def sqs():

    with mock_sqs():

        yield boto3.client('sqs', region_name=AWS_REGION)

This is one of the correct ways of using moto mocks. But only this mock on SQS shows the following warning:

FutureWarning: The sqs client is currently using a deprecated endpoint: eu-central-1.queue.amazonaws.com. In the next minor version this will be moved to sqs.eu-central-1.amazonaws.com. See https://github.com/boto/botocore/issues/2705 for more details.

 Now, our Sonar is considering these warnings as errors, and leads our build to fail. As explained by the issue, the warning comes from the use of a deprecated attribute called sslCommonName. I didn't dive into the moto library, but there might be a way to avoid the use of this attribute. The easisest way to stop using it, is to add this environment variable to our test code:

import os

os.environ['BOTO_DISABLE_COMMONNAME'] = 'true'

This skips completely the use of the deprecated attribute.

AWS: Unblock CloudFormation stacks from UPDATE_ROLLBACK_FAILED state

 In our project, we have lots of AWS accounts. And we are in charge of deploying base resources in each of them. To do this, we use CloudFormation to deploy several stacks. Actually several stacks nested into one master stack.

One problem we have is that our customers do not always update their stacks to the latest version. Also, when a stack fails to update, they sometimes let it rollback, and do not care to ask for a fix. Of course, this summer, there was even less updates, people being on vacation. And also no release, since we felt there would be nobody to deploy it. So we end up this September with a larger release than usual.

The result of all this is that we found ourselves staring at lots of accounts with a stack entering UPDATE_ROLLBACK_FAILED state. The reason? Deprecation. As it happened, AWS decided to deprecate Python 3.6, and also a couple of Policies (AWSConfigRole, AWSCloudTrailReadOnlyAccess). Of course we updated our stacks with the correct values for Python and the replacement policies some time ago. But as the stacks were not always up to date, and there were some issues while updating to the new release, many stacks started to rollback. And since some of the rollbacked values were deprecated, the rollbacks failed.

When you are in that case, you have two choices. The first one is to delete everything and redeploy. We tried it on one account, and it was really painful. Too many dependencies and manual actions. The second one is the one advised to us by the AWS support itself: continuing rolling back. When you continue a rollback, you have the possibilities to skip some resources. In our case, we needed to skip all lambdas using Python 3.6, and all roles using the deprecated policies.

We tried it manually in the AWS Console, and there is one caveat: you can select resources from nested stacks, but not resources from stacks nested into nested stacks. Since we have many accounts to update, and many resources to rollback, we decided to script the whole process.

We thought it will be simple: using Python and boto3, we list all the resources in our stack, recursively entering nested stacks, and filtering all lambdas and roles. We ran into several problems:

• You cannot skip resources that are not in a failed stack
• You cannot skip resources that are not in a failed state
• You cannot skip resources that are in a failed state because CloudFormation cancelled the update
• Once you run rollback with skipped resources, CloudFormation discovers new failing resources, so you have to iterate until all is fine, or the list of resources to skip does not change between two iterations.
• Name of resources in nested stack are <nested_stack_name>.<resource_logical_id>. Even for resources in several level of nested stack, you still use the same pattern, giving only the name of the direct parent stack.
• Waiting for a rollback to complete will throw an exception if the rollback fails.

This is the script that helps turning a stack from UPDATE_ROLLBACK_FAILED to UPDATE_ROLLBACK_COMPLETE:

import boto3

BLOCKABLE_RESOURCES = [

    "AWS::Lambda::Function",

    "AWS::IAM::Role",

]

STACK_NAME = "MyStack"

def get_stack_status(cf_client, stack_name):

    response = cf_client.describe_stacks(StackName=stack_name)

    return response["Stacks"][0]["StackStatus"]

def find_blocking_resources(cf_client, stack_name, parent, resources):

    response = cf_client.describe_stack_resources(StackName=stack_name)

    if (

        parent

        and get_stack_status(cf_client, stack_name) != "UPDATE_ROLLBAK_FAILED"

    ):

        return

    for resource in response["StackResources"]:

        if (

            resource["ResourceType"] == "AWS::CloudFormation::Stack"

            and resource["ResourceStatus"] == "UPDATE_FAILED"

        ):

            nested_name = resource["PhysicalResourceId"].split("/")[1]

            find_blocking_resources(

                cf_client, nested_name, nested_name + ".", resources

            )

        elif (

            resource["ResourceType"] in BLOCKABLE_RESOURCES

            and resource["ResourceStatus"] == "UPDATE_FAILED"

            and resource.get("ResourceStatusReason")

            != "Resource update cancelled"

        ):

            resource.append(parent + resource["LogicalResourceId"])

cf_client = boto3.client("cloudformation")

status = get_stack_status(cf_client, STACK_NAME)

if status != "UPDATE_ROLLBACK_FAILED":

    print("Nothing to unblock. Exiting")

    exit(0)

resources = []

previous_resources = ["DUMMY"]

waiter = cf_client.get_waiter("stack_rollback_complete")

print("Starting unblocking process")

while status != "UPDATE_ROLLBACK_FAILED" and previous_resources != resources:

    previous_resources = resources

    resources = []

    find_blocking_resources(cf_client, STACK_NAME, "", resources)

    print("Skipping ", resources)

    cf_client.continue_update_rollback(

        StackName=STACK_NAME, ResourcesToSkip=resources

    )

    try:

        waiter.wait(StackName=STACK_NAME)

    except Exception as err:

        print(err)

    status = get_stack_status(cf_client, STACK_NAME)

print("Final stack status:", status)

Of course, once this is done, you still have to fix the stacks and run an update.

Modifying AWS resource for testing

 If you are writing unit tests using boto3 and moto, you might want to modify a value in a resource. I had this case, for instance, where I wanted to set the state of an AMI to pending for a test. Using moto, AMIs are always created in the available state. However, if you try to do it this way, it will fail:

ec2 = boto3.resource("ec2")

image = ec2.Image(AMI_ID)

image.state = "pending"

You will end up with this message:

AttributeError: can't set attribute 'state'

It is possible to modify the resource object. However, it will just modify this object instance, the model itself will not be updated. So if you use the resource as a parameter to a function, it will work. But if the function retrieves its data with boto3, it will get the original value.

Here is my solution:

image.meta.data.update({"State": "pending"})

That helped in my case. 

WTF: Case for a Join

 I found this code in our repository:

roles = ""

for i, id in enumerate(account_ids):

    if i > 0:

        roles = roles + ","

        if i == len(account_ids) - 1:

            roles = roles + "\"arn:aws:iam::*:role/" + entity_name + "-" + id + "-admin"

        else:

            roles = roles + "\"arn:aws:iam::*:role/" + entity_name + "-" + id + "-admin\""

    if i == 0:

        if i == len(account_ids) - 1:

            roles = roles + "arn:aws:iam::*:role/" + entity_name + "-" + id + "-admin"

        else:

            roles = roles + "arn:aws:iam::*:role/" + entity_name + "-" + id + "-admin\""

It takes some time to understand what is going on here, but basically, we are building a coma separated list of AWS roles from a list of account IDs.

When you see a condition based on the index of a for loop, you start to feel a very strong code smell.

First, the test for the index being bigger than zero, or equal to 0, is mainly made for handling the coma. But not only. There are then tests repeated, with almost similar codes, to check if we are on the last iteration, all this to discover if we have to add a " character at the beginning or the end of our string.

At the end, it produces a string in the form: role1","role2","role3

The reason that there is no quotation marks at the beginning or the end of the string, is that ultimately, it will be put in a template that is declared with the marks already there, in this form: "__ROLES__".

All this code is quite bad, and the coma and quotation marks handling can all be left to a call to the join function:

roles = '","'.join([f"arn:aws:iam::*:role/{entity_name}-{id}-admin"

    for id in account_ids])

From 13 lines to 1, I kind of like it.

AWS assume role one-liner

 A couple of months ago, I wanted to simplify the way I assume a role in AWS, an operation I perform several times a day. Usually, you would run a command like this one with the AWS CLI:

aws sts assume-role --role-arn $MY_ROLE_ARN --role-session-name test

It would return you a JSON document like this one:

{

    "AssumedRoleUser": {

        "AssumedRoleId": "AROA3XFRBF535PLBIFPI4:s3-access-example",

        "Arn": "arn:aws:sts::123456789012:assumed-role/xaccounts3access/s3-access-example"

    },

    "Credentials": {

        "SecretAccessKey": "9drTJvcXLB89EXAMPLELB8923FB892xMFI",

        "SessionToken": "AQoXdzELDDY//////////wEaoAK1wvxJY12r2IrDFT2IvAzTCn3zHoZ7YNtpiQLF0MqZye/qwjzP2iEXAMPLEbw/m3hsj8VBTkPORGvr9jM5sgP+w9IZWZnU+LWhmg+a5fDi2oTGUYcdg9uexQ4mtCHIHfi4citgqZTgco40Yqr4lIlo4V2b2Dyauk0eYFNebHtYlFVgAUj+7Indz3LU0aTWk1WKIjHmmMCIoTkyYp/k7kUG7moeEYKSitwQIi6Gjn+nyzM+PtoA3685ixzv0R7i5rjQi0YE0lf1oeie3bDiNHncmzosRM6SFiPzSvp6h/32xQuZsjcypmwsPSDtTPYcs0+YN/8BRi2/IcrxSpnWEXAMPLEXSDFTAQAM6Dl9zR0tXoybnlrZIwMLlMi1Kcgo5OytwU=",

        "Expiration": "2016-03-15T00:05:07Z",

        "AccessKeyId": "ASIAJEXAMPLEXEG2JICEA"

    }

}

And then you would need to export environment variables for setting the access key, secret key and session token.

export AWS_ACCESS_KEY_ID="ASIAJEXAMPLEXEG2JICEA"
export AWS_SECRET_ACCESS_KEY="9drTJvcXLB89EXAMPLELB8923FB892xMFI"
export AWS_SESSION_TOKEN="..."

So I looked for a simpler solution and I stumbled upon this StackOverfow question: AWS sts assume role in one command. 

Some suggestions use the very useful JQ utility which allows to retrieve information from JSON documents. But in case of AWS CLI commands, it is normally not necessary, since they all accept the --query option that supports JMESPath syntax. So, as one answer suggested, you can simply use the join built-in command to construct your export command, and let the shell evaluate it. Which is the solution I used for some months now:

eval $(aws sts assume-role \

 --role-arn $MY_ROLE_ARN \

 --role-session-name test \

 --query 'join(``, [`export AWS_ACCESS_KEY_ID=`,

 Credentials.AccessKeyId, ` ; export AWS_SECRET_ACCESS_KEY=`,

 Credentials.SecretAccessKey, `; export AWS_SESSION_TOKEN=`,

 Credentials.SessionToken])' \

 --output text)

This command has been really practical, so I decided to add an entry to my blog about it, so I will always remember where to look for it. So I returned to the StackOverflow site, and I found out that a simpler solution was suggested since. It uses the built-in printf shell function, that I had absolutely no idea existed:

export $(printf "AWS_ACCESS_KEY_ID=%s AWS_SECRET_ACCESS_KEY=%s AWS_SESSION_TOKEN=%s" \

$(aws sts assume-role \

--role-arn $MY_ROLE_ARN \

--role-session-name test \

--query "Credentials.[AccessKeyId,SecretAccessKey,SessionToken]" \

--output text))

I guess you never stop learning.

AWS: Read Timeout when Invoking Lambda

 I have an AWS Lambda that invokes another Lambda synchronously. Nothing was wrong with it until recently, when the other Lambda started performing more tasks and taking more time. Suddenly, the caller Lambda started failing with this weird message:

ReadTimeoutError: Read timeout on endpoint URL: "https://lambda.eu-central-1.amazonaws.com/2015-03-31/functions/MyOtherLambda/invocations"

Looking at the logs of my other Lambda, I could see that it ran fine, although it was executed several times. After some research on the Internet, I found this article from AWS support that explains that when invoking a Lambda synchronously, there is a default timeout of 60 seconds, and 3 retries. This can be configured when creating the client.

In Python using boto3 for instance, you have to use the Config object:

import boto3

from botocore.config import Config

lambda_client = boto3.client("lambda", config=Config(read_timeout=600))

Now, my invoked lambda has 10 minutes to perform its tasks.

Find reasons of failed DynamoDB Transaction with boto3

 When working with Transactions in DynamoDB, you pack several actions into one query. When one of them failed, all the queries are rolled back. But how do you know which one failed? Reading the documentation, you find the following note:

If using Java, DynamoDB lists the cancellation reasons on the CancellationReasons property. This property is not set for other languages. Transaction cancellation reasons are ordered in the order of requested items, if an item has no error it will have NONE code and Null message.

 Why only in Java? I'm working in Python, using the boto3 library. Do I still have a way to access the information? As it turns out, there is a workaround. When I print the message of  my exception, I see the following:

botocore.errorfactory.TransactionCanceledException: An error occurred (TransactionCanceledException) when calling the TransactWriteItems operation: Transaction cancelled, please refer cancellation reasons for specific reasons [None, ConditionalCheckFailed, None]

 The reasons array is printed at the end of the message. If it is already there, why not have it in a more accessible format? From there, it is not difficult to extract it:

import re

def reasons(e):

    m = re.search("\\[(.*)\\]", str(e))

    reasons = m.group(1).split(", ")

    return [None if reason == "None" else reason for reason in reasons]

Now I can use it like that:

import boto3

client = boto3.client("dynamodb")

try:

    client.transact_write_items(

        [

            {

                "ConditionCheck": {... }

            },

            {

                "Put": {

                    "Item": {...},

                    "ConditionExpression": ...

                }

            },

            {

                "Put": {...}

            },

        ]

    )

except client.exceptions.TransactionCanceledException as e:

    if reasons(e)[0] is not None:

        raise MissingItemError()

    raise OperationFailedError()

I hope this array will be added to the exception's response structure as a normal field in a future version. At least, the feature request exists in boto3.

Creating AWS resources in pytest fixtures with moto

 When unit testing functions that access AWS resources, the easiest way is to use the moto library. For instance, I have to test several functions that access an SQS queue, so I proceed this way:

import boto3

from moto import mock_sqs

def create_queue():

    client = boto3.client("sqs")

    queue_url = client.create_queue(QueueName="queue")["QueueUrl"]

    sqs = boto3.resource("sqs")

    return sqs.Queue(queue_url)

@mock_sqs

def test_something()

    queue = create_queue()

    ...

Since I had this create_queue() call at the beginning of most of my test functions, I wanted to make it a fixture. So i tried this way:

import boto3

from moto import mock_sqs

from pytest import fixture

@fixture

@mock_sqs

def queue():

    client = boto3.client("sqs")

    queue_url = client.create_queue(QueueName="queue")["QueueUrl"]

    sqs = boto3.resource("sqs")

    return sqs.Queue(queue_url)

@mock_sqs

def test_something(queue)

    ...

Unfortunately, this raised an error in my test functions stating that the queue does not exist. The reason for it is that the decorator @mock_sqs on the fixture would destroy my SQS queue as soon as it would leave the queue() method.

The solution is simple: do not use the mock as a decorator, but trigger it when the fixture is initializing and destroy it when it terminates. That means using a yield within the fixture to return the queue:

import boto3

from moto import mock_sqs

from pytest import fixture

@fixture

def queue():

    mock_sqs().start()

    client = boto3.client("sqs")

    queue_url = client.create_queue(QueueName="queue")["QueueUrl"]

    sqs = boto3.resource("sqs")

    yield sqs.Queue(queue_url)

    

    mock_sqs().stop()

def test_something(queue)

    ...

Notice that we have to drop the decorator on the test function as well.

Thanks for the answers to this moto issue.