高效并发抓取 UniProt REST API 多个基因查询结果的完整教程-Python教程-PHP中文网

高效并发抓取 UniProt REST API 多个基因查询结果的完整教程

本文介绍如何使用 Python 的 multiprocessing.Pool 替代 concurrent.futures.ThreadPoolExecutor，真正实现 CPU/IO 并行化，显著提升批量请求 UniProt REST API 的速度，并安全解析 Excel 格式响应、合并结果。

本文介绍如何使用 python 的 `multiprocessing.pool` 替代 `concurrent.futures.threadpoolexecutor`，真正实现 cpu/io 并行化，显著提升批量请求 uniprot rest api 的速度，并安全解析 excel 格式响应、合并结果。

在生物信息学数据获取场景中，频繁调用 UniProt REST API（如 https://rest.uniprot.org/uniprotkb/stream）查询多个基因（如从单细胞 RNA-seq 差异表达基因列表）是常见需求。然而，单纯依赖线程池（ThreadPoolExecutor）往往提速有限——这是因为 Python 的 GIL（全局解释器锁）限制了 CPU 密集型任务的并行性，而 HTTP 请求虽属 IO 密集型，但当响应体较大（如 Excel 文件）、解析开销高（pandas.read_excel 本身较重）时，线程竞争和内存拷贝反而成为瓶颈。

此时，多进程（multiprocessing）是更优解：每个进程拥有独立的 Python 解释器和内存空间，可绕过 GIL，真正并行执行网络请求 + Excel 解析 + 数据提取，尤其适合“请求-解析-结构化”这一整条链路。

Wordpress REST API

OpenClaw技能，提供WordPress REST API命令行工具，支持通过原生HTTP处理文章、页面、分类、标签、用户及自定义请求。

下载

以下是一个生产就绪的优化方案，已融合原代码逻辑并增强健壮性：

✅ 核心改进点

使用 multiprocessing.Pool 实现真正的并行请求与解析；
采用 pool.imap_unordered() 流式处理，避免内存积压；
内置异常捕获与重试机制，防止单个失败中断全部流程；
显式控制 chunksize 提升 IPC（进程间通信）效率；
返回结构化元组 (gene_name, first_row_df)，便于后续调试与去重。

? 完整可运行代码示例

import warnings
import pandas as pd
import requests
from io import BytesIO
from multiprocessing import Pool, current_process
from tqdm import tqdm
import time

# 忽略 openpyxl 样式警告（非关键）
warnings.simplefilter("ignore", UserWarning)

def get_UniPQuery_link(gene_name, tax_id="9606", file_format="xlsx"):
    """生成 UniProt REST API 查询链接（精简版，移除冗余参数）"""
    query = f"((gene:{gene_name}) AND (taxonomy_id:{tax_id}))"
    fields = "accession,reviewed,id,protein_name,gene_names," \
             "organism_name,length,ft_intramem,cc_subcellular_location," \
             "ft_topo_dom,ft_transmem,cc_function,lit_doi_id,lit_pubmed_id"
    url = (
        f"https://rest.uniprot.org/uniprotkb/stream?"
        f"fields={fields}&format={file_format}&size=500&query={query}"
    )
    return url

def fetch_gene_data(gene_name, max_retries=3):
    """
    单基因请求 + 解析函数（进程内执行）
    返回: (gene_name, first_row_df) 或 (gene_name, None) 若失败
    """
    url = get_UniPQuery_link(gene_name)

    for attempt in range(max_retries):
        try:
            response = requests.get(url, timeout=30)
            response.raise_for_status()  # 抛出 4xx/5xx 异常

            # 解析 Excel 响应
            df = pd.read_excel(BytesIO(response.content))

            if not df.empty:
                return gene_name, df.iloc[[0]].assign(query_gene=gene_name)
            else:
                return gene_name, None

        except Exception as e:
            if attempt == max_retries - 1:
                print(f"[{current_process().name}] ❌ 失败 {gene_name}: {e}")
                return gene_name, None
            time.sleep(0.5 * (2 ** attempt))  # 指数退避

    return gene_name, None

if __name__ == "__main__":
    # 示例：替换为你的 unique_genes 列表
    unique_genes = ["TP53", "EGFR", "BRAF", "KRAS", "PIK3CA", "AKT1"]

    print(f"▶ 开始并发请求 {len(unique_genes)} 个基因...")
    start_time = time.time()

    # 启动 4 个工作进程（可根据 CPU 核数调整）
    with Pool(processes=4) as pool:
        # imap_unordered + tqdm 实现带进度的流式处理
        results = list(
            tqdm(
                pool.imap_unordered(fetch_gene_data, unique_genes, chunksize=2),
                total=len(unique_genes),
                desc="? 获取 UniProt 数据"
            )
        )

    # 分离成功/失败结果
    valid_dfs = [df for _, df in results if df is not None]
    failed_genes = [name for name, df in results if df is None]

    # 合并所有有效结果
    if valid_dfs:
        final_df = pd.concat(valid_dfs, ignore_index=True)
        final_df.to_excel("uniprot_gene_summary.xlsx", index=False)
        print(f"\n✅ 成功获取 {len(valid_dfs)} 条记录，已保存至 uniprot_gene_summary.xlsx")
    else:
        print("\n⚠️  未获取到任何有效数据，请检查网络或基因名拼写。")

    if failed_genes:
        print(f"⚠️  以下基因请求失败: {failed_genes}")

    elapsed = time.time() - start_time
    print(f"⏱️  总耗时: {elapsed:.2f} 秒 ({elapsed/60:.2f} 分钟)")

⚠️ 关键注意事项

API 限流：UniProt REST API 对未认证用户有请求频率限制（约 10 req/sec）。若批量量极大（>1000 基因），建议：
- 添加 time.sleep(0.1) 在 fetch_gene_data 中（慎用，会降低吞吐）；
- 或改用 UniProt Batch Retrieval 或下载全量数据库本地匹配。
内存管理：Excel 文件较大时（>5MB），pd.read_excel 可能触发 OOM。可改用 engine='openpyxl' + usecols 限定列，或直接解析 TSV（format='tsv' 更轻量）。
Windows 兼容性：务必确保 if __name__ == "__main__": 包裹主逻辑，否则多进程在 Windows 下会递归启动。
结果去重：iloc[[0]] 仅取首条，但某些基因名可能对应多个蛋白（如剪接变体）。如需全量，可移除 .iloc[[0]] 并保留整个 df。

✅ 性能对比小结

方法	理论并行性	实际提速（~100 genes）	适用场景
ThreadPoolExecutor	线程级（受 GIL 制约）	~1.5–2×	轻量 JSON/API 响应
multiprocessing.Pool	进程级（无 GIL）	~3–4×（尤其大 Excel）	本例（含 Excel 解析）
异步（aiohttp + asyncio）	协程级（高并发 IO）	~2.5–3.5×	需改造解析逻辑，学习成本高