python - 使用 async/await 从 python 异步调用 AWS Lambda 函数

Question

我有一个 AWS lambda 函数，需要异步调用(即发即忘)，并在以非阻塞方式完成时获取结果。

loop = asyncio.get_event_loop()

async def f(payload):
    lambda_client = boto3.client('lambda')
    response = lambda_client.invoke(
    FunctionName='FUNC_NAME',
    InvocationType='RequestResponse',
    LogType='Tail',
    Payload=payload,
    Qualifier='$LATEST'
    )
    response_body = response['Payload']
    response_str = response_body.read().decode('utf-8')
    response_dict = eval(response_str)
    return response_dict


async def g():

    payload = json.dumps({
      "test_bucket": "MY_BUCKET",
      "test_key": "my_test_key.csv",
      "testpred_bucket": "MY_BUCKET",
      "testpred_key": "my_test_key_new.csv",
      "problem": "APROBLEM"
    })
    # Pause here and come back to g() when f() is ready
    r = await f(payload)
    print(r)

这可行，但并不能真正达到即发即忘的目的。我知道我需要使用 asyncio.ensure_future 但如果我使用 asyncio.ensure_future(f(payload))，如何捕获函数的返回值f。我是 python async 新手，不清楚。

有人可以建议吗？

Answer 1

• 这就是我在 aws lambda 中所做的，我想做 4 个发布请求，然后收集所有响应。

    loop = asyncio.get_event_loop()

    if loop.is_closed():
        loop = asyncio.new_event_loop()

    #The perform_traces method i do all the post method
    task = loop.create_task(perform_traces(payloads, message, contact_centre))
    unique_match, error = loop.run_until_complete(task)
    loop.close()

在perform_trace方法中，这就是我使用 session 等待的方式

    future_dds_responses = []

    async with aiohttp.ClientSession() as session:

        for payload in payloads:
            future_dds_responses.append(dds_async_trace(session, payload, contact_centre))

        dds_responses, pending = await asyncio.wait(future_dds_responses)

在dds_async_trace中，这就是我使用aiohttp.ClientSession session 完成帖子的方式

        async with session.post(pds_url,
                                data=populated_template_payload,
                                headers=PDS_HEADERS,
                                ssl=ssl_context) as response:
            status_code = response.status