Multiple parallel AWS Lambda invocations

Question

I am trying to execute multiple AWS Lambda invocations with python 3.7.2 and aiobotocore package. Here is my code.

import asyncio
import aiobotocore


async def invoke(payload, session):
    async with session.create_client('lambda', region_name='us-east-1') as client:
        return await client.invoke(FunctionName='MY_FUNCTION', Payload=payload)


def generate_invocations(payloads, session):
    for payload in payloads:
        yield invoke(payload, session)


def invoke_all(payloads):
    loop = asyncio.get_event_loop()

    async def wrapped():
        session = aiobotocore.get_session(loop=loop)
        invocations = generate_invocations(payloads, session)
        return await asyncio.gather(*invocations)

    return loop.run_until_complete(wrapped())


def main():
    payloads_list = []  # MY PAYLOADS LIST 
    lambda_responses = invoke_all(payloads_list)
    print(lambda_responses)


if __name__ == '__main__':
    main()

The code runs really fast (for 10 payloads around 1 sec instead of 15 by using boto3 lambda client invocations) but I have two issues:

1) The elements in lambda_responses include 'Payload' key which value is of type aiobotocore.response.StreamingBody. When I try value.read() I receive "coroutine object StreamingBody.read" and I think there some problem in my code. I can receive desired response by "json.loads(json.loads(r['Payload']._buffer.pop())['body'])" but what is the proper way to get it.

2) In rare cases the "Payload" in one of responses has empty buffer. How can I ensure that invoke_all function returns non empty responses? Is it correct usage of aiobotocore?

I am new to python 3 and async functionality. Inspired by examples of aiobotocore documentation and Mathew Marcus blog.

Thanks!

Roelant · Accepted Answer · 2021-03-21 07:53:01Z

7

The elements in lambda_responses include 'Payload' key which value is of type aiobotocore.response.StreamingBody. When I try value.read() I receive "coroutine object StreamingBody.read"

This means the read() coroutine is meant to be awaited, which you should do while still in the event loop. For example, you could change the invoke coroutine to also read the response:

async def invoke(payload, session):
    async with session.create_client('lambda', region_name='us-east-1') as client:
        resp = await client.invoke(FunctionName='MY_FUNCTION', Payload=payload)
        payload = await resp['Payload'].read()
        return payload  # or assemble a dict with relevant parts

In rare cases the "Payload" in one of responses has empty buffer.

This is likely because you are accessing the buffer before actually reading the contents. In some cases the info arrives soon enough that you find it in the internal buffer anyway, but sometimes you have to wait for it. Using a public method such as read() ensures that you're correctly using the API. The _buffer property, on the other hand, begins with an underscore, which denotes that it is an implementation detail not meant to be accessed directly.

edited Mar 21, 2021 at 7:53

Roelant

5,2295 gold badges43 silver badges77 bronze badges

answered Feb 26, 2019 at 23:18

user4815162342

159k22 gold badges351 silver badges419 bronze badges

Sign up to request clarification or add additional context in comments.

1 Comment

maksim Over a year ago

Thanks! Works great.

Collectives™ on Stack Overflow

Multiple parallel AWS Lambda invocations

1 Answer 1

1 Comment

Your Answer

Hot Network Questions

Collectives™ on Stack Overflow

1 Answer 1

1 Comment

Your Answer

Sign up or log in

Post as a guest

Related