AWS异步SQS队列代码报错：AttributeError及协程未等待问题求助

问题代码 #

import asyncio
import os
import aioboto3
from typing import Dict, List

from src.data.models.post_webinar.lead_classes import OutreachProspect, SalesforceCampaignMember

class AsyncCommunicator:

    async def async_signal_flow(self, campaign_members: List[SalesforceCampaignMember]):
        tasks = []
        for cm in campaign_members:
            cm = cm.as_json()
            task = asyncio.ensure_future(self._send_signal_to_sqs_queue("SIGNAL_PROCESSING_QUEUE", cm))
        tasks.append(task)
        await asyncio.gather(*tasks)

    async def async_sequence(self, prospects: List[OutreachProspect], webinar_uuid: str):
        tasks = []
        for prospect in prospects:
            prospect = prospect.as_json()
            prospect["webinar_uuid"] = webinar_uuid
            task = asyncio.ensure_future(self._send_signal_to_sqs_queue("SEQUENCE_PROCESSING_QUEUE", prospect))
            tasks.append(task)
        await asyncio.gather(*tasks)
    
    @staticmethod
    async def _send_signal_to_sqs_queue(queue_name: str, serialized_object: Dict) -> None:
        session = aioboto3.Session()
        async with session.resource('sqs') as sqs_client:
            queue = await sqs_client.get_queue_by_name(
                QueueName=os.environ[queue_name]
            )
            await queue.send_message(
                MessageBody=str(serialized_object)
        )

报错信息 #

{
    "success": false,
    "error": "AttributeError(\"'ClientCreatorContext' object has no attribute 'meta'\")"
}

sys:1: RuntimeWarning: coroutine 'AioSession._create_client' was never awaited
RuntimeWarning: Enable tracemalloc to get the object allocation traceback

问题分析与修复 #

• aioboto3资源使用错误：aioboto3.Session.resource() 返回的是协程对象，不能直接用async with包裹，需要先await获取资源实例。这是触发AttributeError和协程未等待警告的核心原因。
• 任务添加缩进错误：async_signal_flow方法中，tasks.append(task)的缩进位置错误，导致循环内仅最后一个任务被添加到列表，前面的任务全部丢失。

修复后的完整代码 #

import asyncio
import os
import aioboto3
from typing import Dict, List

from src.data.models.post_webinar.lead_classes import OutreachProspect, SalesforceCampaignMember

class AsyncCommunicator:

    async def async_signal_flow(self, campaign_members: List[SalesforceCampaignMember]):
        tasks = []
        for cm in campaign_members:
            cm = cm.as_json()
            task = asyncio.create_task(self._send_signal_to_sqs_queue("SIGNAL_PROCESSING_QUEUE", cm))
            tasks.append(task)  # 修复缩进，确保每个任务都被添加到列表
        await asyncio.gather(*tasks)

    async def async_sequence(self, prospects: List[OutreachProspect], webinar_uuid: str):
        tasks = []
        for prospect in prospects:
            prospect = prospect.as_json()
            prospect["webinar_uuid"] = webinar_uuid
            task = asyncio.create_task(self._send_signal_to_sqs_queue("SEQUENCE_PROCESSING_QUEUE", prospect))
            tasks.append(task)
        await asyncio.gather(*tasks)
    
    @staticmethod
    async def _send_signal_to_sqs_queue(queue_name: str, serialized_object: Dict) -> None:
        session = aioboto3.Session()
        # 先await获取sqs资源实例，再使用async with
        sqs_resource = await session.resource('sqs')
        async with sqs_resource as sqs_client:
            queue = await sqs_client.get_queue_by_name(
                QueueName=os.environ[queue_name]
            )
            await queue.send_message(
                MessageBody=str(serialized_object)
            )

额外优化建议 #

• 替换asyncio.ensure_future为asyncio.create_task，这是Python 3.7+推荐的异步任务创建方式，语义更清晰。
• 建议为os.environ[queue_name]添加异常处理，避免环境变量不存在时直接抛出KeyError。

内容的提问来源于stack exchange，提问作者hughmcmc