如何使用RxNorm API返回的rxcui调用RxClass API提取药品分类

现有代码与需求 #

需要基于NDC编码调用RxNorm API提取药品分类，已完成基础信息查询的Python代码如下：

#!/usr/bin/python
# 依赖安装：pip install requests simplejson
import os
import sys
import requests
import simplejson as json

def connectionCheck():
    url = 'http://rxnav.nlm.nih.gov/REST/version'
    header = {'Accept': 'application/json'}
    getCheck = requests.get(url, headers=header)
    if getCheck.status_code != requests.codes.ok:
        response = "RXNorm server response error. Response code: %s" % getCheck.status_code
    else:
        response = "Connection check complete. RXNorm online. Response code: %s" % getCheck.status_code
    return response

def rxNorm(ndc):
    # ndc值从主程序传入，格式为NDC编码数组
    if ndc[0] is None:
        return {"rxcui": "", "rxtty": "", "rxstring": ""}
    else:
        try:
            baseurl = 'http://rxnav.nlm.nih.gov/REST/'

            # NDC查RXCUI接口路径
            ndcSearch = 'rxcui?idtype=NDC&id='
            # 概念属性查询接口路径
            rxPropSearch = 'rxcui/'
            rxttySearch = '/property?propName=TTY'
            rxstringSearch = '/property?propName=RxNorm%20Name'

            header = {'Accept': 'application/json'}
            def getTTY(rxCUI):
                getTTY = requests.get(baseurl+rxPropSearch+rxCUI+rxttySearch, headers=header)
                ttyJSON = json.loads(getTTY.text, encoding="utf-8")
                return ttyJSON['propConceptGroup']['propConcept'][0]['propValue']

            def getSTRING(rxCUI):
                getString = requests.get(baseurl+rxPropSearch+rxCUI+rxstringSearch, headers=header)
                stringJSON = json.loads(getString.text, encoding="utf-8")
                return stringJSON['propConceptGroup']['propConcept'][0]['propValue']

            # 遍历传入的NDC列表查询有效结果
            for item in ndc:
                getRXCUI = requests.get(baseurl+ndcSearch+item, headers=header)
                if getRXCUI.status_code != requests.codes.ok:
                    print ("RXNorm server response error. Response code: %s" % getRXCUI.status_code)
                rxcuiJSON = json.loads(getRXCUI.text, encoding="utf-8")
                try:
                    if rxcuiJSON['idGroup']['rxnormId']:
                        rxCUI = rxcuiJSON['idGroup']['rxnormId'][0]
                        rxTTY = getTTY(rxCUI)
                        rxSTRING = getSTRING(rxCUI)
                        return {"rxcui": rxCUI, "rxtty": rxTTY, "rxstring": rxSTRING}
                except:
                    if item == ndc[-1]:
                        return {"rxcui": "", "rxtty": "", "rxstring": ""}
                    pass
        except:
            sys.exit("RXNorm connection error")

使用测试NDC编码调用上述函数：

dataTest=rxNorm(['69238131109'])
print(dataTest)

得到返回结果：

{'rxcui': '483448', 'rxtty': 'SCD', 'rxstring': 'pregabalin 50 MG Oral Capsule'}

当前需要基于返回的rxcui值（示例中为483448），调用RxClass API的getClassByRxNormDrugId接口获取对应药品分类，需要明确该接口的正确调用方式。

接口调用实现 #

接口规则说明 #

• 接口请求路径为http://rxnav.nlm.nih.gov/REST/rxclass/class/byRxcui
• 核心必填参数为rxcui，即从RxNorm拿到的药品概念ID
• 可选参数classSources可指定需要查询的分类体系，常用值包括ATC（解剖治疗化学分类系统）、MED-RT（药物参考术语）等，多个分类源用英文逗号分隔，不传则默认返回所有分类体系下的结果
• 路径末尾加.json后缀即可返回JSON格式响应，和现有代码的请求头规则完全兼容

可复用查询函数 #

直接在现有代码中新增如下函数即可实现分类查询：

def get_rxclass_by_rxcui(rxcui, class_sources=None):
    """
    基于rxcui查询药品分类
    :param rxcui: RxNorm概念ID
    :param class_sources: 指定分类源，例如"ATC,MED-RT"，不传返回所有分类
    :return: 结构化分类结果列表
    """
    if not rxcui:
        return []
    base_url = "http://rxnav.nlm.nih.gov/REST/rxclass/class/byRxcui.json"
    params = {"rxcui": rxcui}
    if class_sources:
        params["classSources"] = class_sources
    header = {'Accept': 'application/json'}
    
    try:
        resp = requests.get(base_url, params=params, headers=header)
        if resp.status_code != 200:
            print(f"RxClass接口响应错误，状态码：{resp.status_code}")
            return []
        res_json = json.loads(resp.text, encoding="utf-8")
        # 解析分类结果
        class_list = []
        rxclass_result = res_json.get("rxclassDrugInfoList", {}).get("rxclassDrugInfo", [])
        for item in rxclass_result:
            class_info = item.get("rxclassMinConceptItem", {})
            class_list.append({
                "class_id": class_info.get("classId", ""),
                "class_name": class_info.get("className", ""),
                "class_type": class_info.get("classType", ""),
                "class_source": class_info.get("classSource", "")
            })
        # 去重后返回
        return [dict(t) for t in {tuple(d.items()) for d in class_list}]
    except Exception as e:
        print(f"查询RxClass出错：{str(e)}")
        return []

调用示例 #

拿到rxNorm函数返回的结果后，直接传入rxcui即可查询：

# 先查询基础信息
dataTest=rxNorm(['69238131109'])
print("基础药品信息：", dataTest)
# 查询分类，示例只查ATC分类，不需要可以不传class_sources参数
drug_class = get_rxclass_by_rxcui(dataTest["rxcui"], class_sources="ATC")
print("药品分类结果：", drug_class)

示例中rxcui=483448（普瑞巴林50mg口服胶囊）的ATC分类返回结果为：

[
    {
        'class_id': 'N03AX16',
        'class_name': 'pregabalin',
        'class_type': 'IN',
        'class_source': 'ATC'
    },
    {
        'class_id': 'N03AX',
        'class_name': 'Other antiepileptics',
        'class_type': 'TC',
        'class_source': 'ATC'
    }
]

注意事项 #

• 部分老药、小众药可能没有对应分类信息，函数已经做空值兼容，不会抛出键错误
• 如果需要多维度分类，直接在class_sources参数里拼接需要的分类源即可，例如class_sources="ATC,MED-RT,VA"
• RxNorm官方接口有请求频率限制，建议批量查询时加100-200ms的延时，避免被临时封禁

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