Python中不使用aiohttp如何正确发起异步请求？（HubSpot场景）

在仅内置requests包的HubSpot自定义代码环境中，你有两种可行方式实现高效的异步请求，无需额外安装依赖：

方案1：优化线程池式异步（代码简洁，复用requests能力） #

虽然本质还是基于线程池，但相比asyncio.to_thread，通过显式配置ThreadPoolExecutor可以更精准控制资源使用，适配HubSpot的内存和时长限制。这种方式能直接复用requests的所有特性（如请求头、认证、超时处理等），代码改动极小。

示例代码：

import asyncio
import requests
from concurrent.futures import ThreadPoolExecutor

# 根据API速率限制和内存情况调整线程数，建议3-5
MAX_WORKERS = 4

async def async_request(url, **kwargs):
    with ThreadPoolExecutor(max_workers=MAX_WORKERS) as executor:
        loop = asyncio.get_event_loop()
        # 将requests.get提交到线程池执行
        response = await loop.run_in_executor(executor, requests.get, url, **kwargs)
        return response

# 批量获取记录示例
async def sync_records(url_list):
    # 创建所有请求任务
    tasks = [async_request(url, headers={"Authorization": "Bearer YOUR_TOKEN"}) for url in url_list]
    # 并发执行并收集结果
    responses = await asyncio.gather(*tasks)
    # 过滤有效响应并解析
    return [resp.json() for resp in responses if resp.status_code == 200]

方案2：纯异步IO实现（无多线程，低内存开销） #

如果需要严格的异步非阻塞实现，可以基于Python标准库的asyncio和套接字功能，手动实现HTTP请求逻辑。这种方式不依赖线程池，内存占用更低，但需要自己处理HTTP协议的细节（如HTTPS、请求头、响应解析等）。

示例代码：

import asyncio
from urllib.parse import urlparse
import json

async def async_http_get(url, headers=None):
    parsed_url = urlparse(url)
    host = parsed_url.netloc
    path = parsed_url.path or "/"
    if parsed_url.query:
        path += f"?{parsed_url.query}"
    
    # 建立异步SSL连接
    reader, writer = await asyncio.open_connection(host, 443, ssl=True)
    
    # 构造HTTP请求头
    request_headers = [
        f"GET {path} HTTP/1.1",
        f"Host: {host}",
        "Connection: close",
    ]
    if headers:
        for key, value in headers.items():
            request_headers.append(f"{key}: {value}")
    request_headers.append("\r\n")
    request = "\r\n".join(request_headers).encode("utf-8")
    
    # 发送请求
    writer.write(request)
    await writer.drain()
    
    # 读取响应
    response_data = await reader.read()
    writer.close()
    await writer.wait_closed()
    
    # 解析响应
    header_part, body_part = response_data.split(b"\r\n\r\n", 1)
    status_line = header_part.split(b"\r\n")[0].decode("utf-8")
    status_code = int(status_line.split()[1])
    
    if status_code == 200:
        return json.loads(body_part)
    return None

# 批量请求示例
async def fetch_multiple_records(url_list, auth_headers):
    tasks = [async_http_get(url, headers=auth_headers) for url in url_list]
    return await asyncio.gather(*tasks)

方案选择建议 #

• 若你的请求逻辑依赖requests的高级特性（如自动重定向、会话保持、复杂认证），优先选方案1，代码维护成本低，且在HubSpot的资源限制下足够高效。
• 若你需要极致的内存效率或避免线程池开销，可选择方案2，但需自行处理HTTP协议的细节（如超时、错误重试等）。

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