AWS Lambda/Asyncio：协程从未被等待

Question

目前正在尝试将报告自动化工具上传到 AWS Lambda，该工具使用 asyncio 创建从 API 收集数据的多个线程。以下是该问题的相关代码：

def campaign_report(config):
    print('Setting Campaign Report Details . . .')
    futures = []

    # fields determine what data to pull from the API
    fields = [
        'campaign_name',
        'actions',
        'impressions',
        'spend'
    ]

    # params determines what type of data to pull from the API
    params = {
        'limit': '10000',
        'date_preset': 'last_30d',
        'level': 'campaign',
        'breakdowns': ['device_platform']
    }

    # general columns that we want to get from insight that have a single layer of data
    general_columns = {
        'campaign_name': 'campaign_name',
        'impressions': 'impressions',
        'spend': 'spend',
        'device_platform': 'device_platform'
    }

    # action columns that we want to get from insight['actions'] that have a multi-layer of data
    # format is action.action_type_returned_from_api: column_name
    action_columns = {
        'action.link_click': 'link_click',
        'action.like': 'like',
        'action.app_install': 'app_install',
        'action.landing_page_view': 'landing_page_view'
    }

    # action videos that we want to get from insight videos that have a multi-layer of data
    # format is action_video_returned_from_api: db_column_name
    action_video_columns = {}

    loop = asyncio.get_event_loop()
    loop.run_until_complete(handle_report(config, fields, params))
    loop.close()

    print('Configuring Data . . .')
    configured_report = CampaignReport().get_insights_values(
        insights=data,
        general_columns=general_columns,
        action_video_columns=action_video_columns,
        action_columns=action_columns
    )

    return {'campaigns': configured_report}


async def handle_report(config, fields, params):
    with concurrent.futures.ThreadPoolExecutor(max_workers=20) as executor:
        loop = asyncio.get_event_loop()
        futures = [
            loop.run_in_executor(executor, fetch_report, account['id'], account['name'], fields, params)
            for account in config['accounts']
        ]

        for response in await asyncio.gather(*futures):
            global data
            print('Exporting %s Data . . .' % response['name'])
            data.append(response['name'])
            data = data + list(response['report'])


def fetch_report(account_id, account_name, fields, params):
    print('Fetching %s Report Data . . .' % account_name)
    report = CampaignReport().get_insights_async(
        account_id,
        account_name,
        fields=fields,
        params=params
    )

    return {'name': account_name, 'report': report}

首先，调用campaign_report() 来构建报告详细信息并实例化异步线程池执行器。执行器根据需要多少帐户创建一个池，并且应该并行执行该池。完成每个报告后，数据将映射到更大的列表并从那里进行配置。

我将以此作为开头，这在本地完美运行，没有明显的问题/错误。但是，一旦我将它上传到 AWS Lambda，我就会收到以下错误：

START RequestId: 977dfaeb-cafa-11e7-9e9e-fdf67b36edf2 Version: $LATEST
Setting Campaign Report Details . . .
A Future or coroutine is required
**** Program Execution Complete --- 1.1237462361653646e-05 minutes ****
A Future or coroutine is required: TypeError
Traceback (most recent call last):
  File "/var/task/lambda_helper.py", line 206, in lambda_handler
    raise e
  File "/var/task/lambda_helper.py", line 198, in lambda_handler
    report = campaign_report(config)
  File "/var/task/lambda_helper.py", line 77, in campaign_report
    loop.run_until_complete(handle_report(config, fields, params))
  File "/var/task/asyncio/base_events.py", line 296, in run_until_complete
    future = tasks.async(future, loop=self)
  File "/var/task/asyncio/tasks.py", line 516, in async
    raise TypeError('A Future or coroutine is required')
TypeError: A Future or coroutine is required

/var/runtime/awslambda/bootstrap.py:264: RuntimeWarning: coroutine 'handle_report' was never awaited
  errortype = "unhandled"
END RequestId: 977dfaeb-cafa-11e7-9e9e-fdf67b36edf2
REPORT RequestId: 977dfaeb-cafa-11e7-9e9e-fdf67b36edf2  Duration: 2.40 ms

它声称除了 TypeError 之外没有声明 await。当我查看我的代码时，我似乎无法找到我在设置 asyncio 时出错的地方。

这里的任何帮助都会很棒！

Answer 1

您的代码适用于 Python >= 3.5.1，请确保您运行的是正确的版本。