如何将aiohttp异步HTTP请求响应与对应contactid关联？

要把响应和对应的contactid绑定起来，核心思路是在异步请求的整个流程中携带contactid，让每个请求的返回结果都包含对应的标识符。下面是具体的代码修改方案：

1. 修改异步请求函数，携带并返回contactid #

首先调整__fetch和__bound_fetch函数，让它们接收contactid参数，并返回包含contactid、响应内容和错误信息的元组，方便后续处理异常：

async def __fetch(self, contactid, url, params, session):
    try:
        async with session.get(url, params=params) as response:
            output = await response.read()
            # 返回(contactid, 响应内容, 错误信息)，错误信息为None表示请求成功
            return (contactid, output, None)
    except Exception as e:
        # 请求失败时返回contactid和错误详情
        return (contactid, None, str(e))

async def __bound_fetch(self, sem, contactid, url, params, session):
    # 带信号量的请求函数，同步携带contactid参数
    async with sem:
        return await self.__fetch(contactid, url, params, session)

2. 调整批量请求的ID与URL配对逻辑 #

在get_contact_properties函数中，不要只生成孤立的URL列表，而是生成(contactid, 对应URL)的元组列表，确保每个请求都能关联到原始ID：

def get_contact_properties(self, office_name, api_key, ids, chunk_size=100, **params):
    properties_pages = []
    batch = 0
    while True:
        # 修正原代码的切片错误：用Python标准列表切片避免ID丢失
        chunk_ids = ids[batch * chunk_size : min((batch + 1) * chunk_size, len(ids))]
        # 生成contactid和对应请求URL的配对列表
        id_url_pairs = [
            (contactid, f"{self.__get_base_url()}/contacts/{contactid}/properties?api_key={api_key}")
            for contactid in chunk_ids
        ]
        # 调用批量请求函数，传入配对列表
        responses_with_ids = self.get_responses(id_url_pairs, self.get_office_token(office_name), chunk_size)
        
        # 处理响应，关联contactid与解析后的结果
        valid_responses = []
        for contactid, response_raw, error in responses_with_ids:
            if error:
                print(f"请求contactid {contactid}失败: {error}")
                continue
            try:
                response_json = json.loads(response_raw)
                # 复用你原有的响应校验逻辑，这里假设__is_valid是校验方法
                if self.__is_valid_contact_property_response(response_json):
                    valid_responses.append({"contactid": contactid, "properties": response_json})
            except Exception as e:
                print(f"解析contactid {contactid}的响应失败: {e}")
        
        properties_pages.append(valid_responses)
        # 判断是否到达最后一页：当前批次的ID数量小于chunk_size则停止
        if len(chunk_ids) < chunk_size:
            break
        batch += 1
    return properties_pages

3. 调整get_responses的任务创建逻辑 #

假设你的get_responses函数负责创建并执行异步任务，需要修改它来处理id_url_pairs，为每个配对创建携带contactid的任务：

def get_responses(self, id_url_pairs, token, chunk_size):
    import asyncio
    sem = asyncio.Semaphore(chunk_size)
    async def main():
        async with aiohttp.ClientSession(headers={"Authorization": f"Bearer {token}"}) as session:
            # 为每个(contactid, url)配对创建异步任务
            tasks = [
                self.__bound_fetch(sem, contactid, url, params=self.params, session=session)
                for contactid, url in id_url_pairs
            ]
            # 等待所有任务完成，返回带ID的结果列表
            return await asyncio.gather(*tasks)
    # 运行异步主函数并返回结果
    return asyncio.run(main())

关键细节说明 #

• 修复了原代码中chunk_ids的切片错误，避免漏掉第一个ID或超出列表范围。
• 每个请求结果都绑定了对应的contactid，后续无论是数据存储还是异常排查，都能直接关联到目标联系人。
• 加入了全流程的异常捕获，即使单个请求失败或解析出错，也能定位到具体的contactid，不会影响其他请求的处理。

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