Python实战：5分钟搞定PubChem API批量查询化合物属性（附完整代码）

Python实战：5分钟高效批量查询PubChem化合物属性的完整指南

在药物研发和化学分析领域，快速获取大量化合物的精确属性数据是每个科研人员的基础需求。传统的手动查询方式不仅效率低下，还容易出错。PubChem作为全球最大的化学数据库之一，其PUG REST API为开发者提供了强大的数据获取能力。本文将带你从零开始，用Python构建一个高性能的批量查询工具，实现每分钟处理上千条化合物记录的效率。

1. 环境准备与API基础

在开始编码前，我们需要确保开发环境就绪。不同于简单的单次请求，批量查询需要考虑网络延迟、数据解析和错误处理等多方面因素。

核心依赖安装：

pip install requests pandas tqdm

PubChem PUG REST API支持多种查询模式，对于批量操作，重点关注以下端点：

• 属性获取：/property/{properties}/{format}
• 同义词获取：/synonyms/{format}
• 批量CID处理：支持逗号分隔的CID列表

注意：PubChem API对未认证用户有每秒3-5次的请求限制，批量查询时需要合理控制请求频率

关键参数说明：

2. 高效批量查询架构设计

要实现高性能的批量查询，我们需要解决三个核心问题：请求分块、错误处理和进度监控。以下是一个经过优化的类结构：

import requests from tqdm import tqdm import pandas as pd from time import sleep class PubChemBatchQuery: def __init__(self, chunk_size=100, delay=0.2): self.base_url = "https://pubchem.ncbi.nlm.nih.gov/rest/pug/compound/cid" self.chunk_size = chunk_size # 每批处理的CID数量 self.delay = delay # 请求间隔防止限流 self.default_props = [ 'MolecularFormula', 'MolecularWeight', 'IsomericSMILES', 'IUPACName', 'XLogP' ] def _chunk_cids(self, cid_list): """将CID列表分块处理""" for i in range(0, len(cid_list), self.chunk_size): yield cid_list[i:i + self.chunk_size]

性能优化关键点：

• 使用请求会话保持连接
• 实现自动重试机制
• 支持进度条显示
• 内存友好的流式处理

3. 完整实现与错误处理

下面给出完整的批量查询实现，包含健壮的错误处理机制：

def query_compounds(self, cid_list, properties=None): """批量查询化合物属性""" properties = properties or self.default_props all_results = [] with requests.Session() as session: for chunk in tqdm(list(self._chunk_cids(cid_list)), desc="Processing"): try: cid_str = ','.join(map(str, chunk)) prop_str = ','.join(properties) url = f"{self.base_url}/{cid_str}/property/{prop_str}/JSON" response = session.get(url) response.raise_for_status() data = response.json() all_results.extend(data['PropertyTable']['Properties']) sleep(self.delay) except requests.exceptions.RequestException as e: print(f"Error processing CIDs {chunk}: {str(e)}") # 失败时尝试单条查询 for single_cid in chunk: try: single_url = f"{self.base_url}/{single_cid}/property/{prop_str}/JSON" single_resp = session.get(single_url) single_data = single_resp.json() all_results.extend(single_data['PropertyTable']['Properties']) sleep(self.delay) except: print(f"Failed to query CID {single_cid}") return pd.DataFrame(all_results)

常见错误及解决方案：

4. 实战应用与数据导出

将查询结果导出为多种格式，便于后续分析：

def export_results(self, df, output_file): """导出查询结果""" if output_file.endswith('.csv'): df.to_csv(output_file, index=False) elif output_file.endswith('.xlsx'): df.to_excel(output_file, index=False) elif output_file.endswith('.json'): df.to_json(output_file, orient='records') else: raise ValueError("Unsupported file format") # 使用示例 if __name__ == "__main__": query_tool = PubChemBatchQuery(chunk_size=50) # 示例CID列表 test_cids = [2244, 2245, 1983, 1988, 2005, 2010] # 自定义需要查询的属性 custom_props = ['MolecularWeight', 'IsomericSMILES', 'XLogP'] # 执行查询 results = query_tool.query_compounds(test_cids, properties=custom_props) # 导出结果 query_tool.export_results(results, 'compound_data.xlsx')

高级技巧：

1. 属性组合查询：一次性获取多种属性减少请求次数
2. 异步请求：使用aiohttp进一步提升性能
3. 结果缓存：避免重复查询相同CID
4. 属性映射表：扩展更多可用属性

# 扩展属性映射表示例 PROPERTY_MAP = { 'formula': 'MolecularFormula', 'weight': 'MolecularWeight', 'smiles': 'IsomericSMILES', 'iupac': 'IUPACName', 'logp': 'XLogP', 'h_bond_donor': 'HBondDonorCount', 'h_bond_acceptor': 'HBondAcceptorCount' }

5. 性能对比与优化建议

通过实际测试对比不同参数下的查询效率：

根据测试结果，推荐以下优化策略：

• 实验室环境：使用50-100的分块大小配合0.2-0.3秒间隔
• 生产环境：实现自动重试和动态调整机制
• 超大规模查询：考虑使用PubChem的FTP批量下载服务

对于需要更高性能的场景，可以尝试异步IO实现：

import aiohttp import asyncio async def async_query(cid_chunk, properties, session): url = f"https://pubchem.ncbi.nlm.nih.gov/rest/pug/compound/cid/{','.join(cid_chunk)}/property/{','.join(properties)}/JSON" async with session.get(url) as response: data = await response.json() return data['PropertyTable']['Properties']