如何优化Python中用PubChemPy将CID转换为ChEMBL ID的代码？

1. 用PubChemPy批量查询并指定所需字段 #

当前代码逐个请求CID且下载完整化合物数据是主要耗时点，改用批量查询并仅请求所需字段可大幅提升效率：

from pubchempy import get_compounds

# 批量获取CID对应的化合物，仅请求synonyms字段
compounds = get_compounds(cids, namespace='cid', properties='synonyms')
chembl_values = []
for compound in compounds:
    if compound.synonyms:
        chembl_id = next((s for s in compound.synonyms if s.startswith('CHEMBL')), None)
        if chembl_id:
            chembl_values.append(chembl_id)

通过properties='synonyms'指定仅返回同义词数据，避免下载冗余信息，同时批量请求减少了HTTP连接建立的开销。

2. 直接调用PubChem REST API（精准高效） #

利用PubChem的交叉引用（XRefs）API，可直接获取CID对应的ChEMBL ID，无需遍历同义词：

import requests

def get_chembl_ids(cids):
    # 拆分CID为批次，避免单次请求过大触发限流
    batch_size = 100
    chembl_map = {}
    
    for i in range(0, len(cids), batch_size):
        batch = cids[i:i+batch_size]
        cid_str = ','.join(map(str, batch))
        url = f'https://pubchem.ncbi.nlm.nih.gov/rest/pug/compound/cid/{cid_str}/xrefs/ChEMBL/json'
        
        response = requests.get(url)
        data = response.json()
        
        if 'InformationList' in data and 'Information' in data['InformationList']:
            for info in data['InformationList']['Information']:
                chembl_map[info['CID']] = info['XRefs'][0]
    
    # 按原CID顺序返回结果，过滤无匹配项
    return [chembl_map.get(cid) for cid in cids if chembl_map.get(cid) is not None]

# 使用示例
chembl_values = get_chembl_ids(cids)

该API直接返回官方交叉引用的ChEMBL ID，结果更精准，响应数据量更小，速度远快于同义词遍历方式。

3. 异步请求并行处理 #

针对大规模CID列表，用异步请求并行发送查询，进一步压缩总耗时：

import aiohttp
import asyncio

async def fetch_chembl(session, cid):
    url = f'https://pubchem.ncbi.nlm.nih.gov/rest/pug/compound/cid/{cid}/xrefs/ChEMBL/json'
    async with session.get(url) as response:
        data = await response.json()
        if 'InformationList' in data and 'Information' in data['InformationList']:
            return data['InformationList']['Information'][0]['XRefs'][0]
        return None

async def get_chembl_ids_async(cids):
    async with aiohttp.ClientSession() as session:
        tasks = [fetch_chembl(session, cid) for cid in cids]
        results = await asyncio.gather(*tasks)
        return [res for res in results if res is not None]

# 使用示例
chembl_values = asyncio.run(get_chembl_ids_async(cids))

异步请求可同时处理多个CID查询，避免逐个请求的等待时间，适合处理数百个以上CID的场景。

额外优化建议 #

• 添加请求重试：用tenacity库实现网络异常自动重试，避免单次请求失败中断流程。
• 控制批次大小：PubChem API有请求频率限制，批量请求时每批次保持在50-100个CID为宜，防止被限流。

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