benchling-integration

Benchling 集成——R&D 平台 SDK

概述

Benchling 是一个用于生命科学研发的云平台。Python SDK 提供对注册库实体（DNA、蛋白质）、库存、电子实验室笔记本和工作流的程序化访问。所有操作都需要 Benchling 租户 URL 和 API 密钥或 OAuth 凭据。

适用场景

• 在 Benchling 注册库中创建、更新或查询生物序列（DNA、RNA、蛋白质）
• 自动执行库存操作（容器、盒子、位置、样品转移）
• 以编程方式创建或查询电子实验室笔记本（ELN）条目
• 构建工作流自动化（创建任务、更新状态、批量操作）
• 将 FASTA 文件或电子表格中的实体批量导入 Benchling
• 将 Benchling 数据导出为 CSV 或外部数据库，以供分析
• 通过事件驱动集成，将 Benchling 与外部系统同步
• 对于本地序列分析（BLAST、比对），请改用 biopython
• 对于化合物数据库，请改用 pubchem-compound-search

先决条件

pip install benchling-sdk

身份验证设置：从 Benchling Profile Settings 获取 API 密钥。将其安全地存储在环境变量中，切勿提交到版本控制系统。

import os
from benchling_sdk.benchling import Benchling
from benchling_sdk.auth.api_key_auth import ApiKeyAuth

benchling = Benchling(
    url="https://your-tenant.benchling.com",
    auth_method=ApiKeyAuth(os.environ["BENCHLING_API_KEY"])
)

OAuth（用于多用户应用）：

from benchling_sdk.auth.client_credentials_oauth2 import ClientCredentialsOAuth2

benchling = Benchling(
    url="https://your-tenant.benchling.com",
    auth_method=ClientCredentialsOAuth2(
        client_id=os.environ["BENCHLING_CLIENT_ID"],
        client_secret=os.environ["BENCHLING_CLIENT_SECRET"]
    )
)

API 速率限制：Benchling 对每个租户实施速率限制。SDK 会在收到 429 响应时自动采用指数退避重试（默认最多重试 5 次）。对于批量操作，请在每批之间添加 time.sleep(0.5)。

快速开始

from benchling_sdk.benchling import Benchling
from benchling_sdk.auth.api_key_auth import ApiKeyAuth
from benchling_sdk.models import DnaSequenceCreate
import os

benchling = Benchling(
    url="https://your-tenant.benchling.com",
    auth_method=ApiKeyAuth(os.environ["BENCHLING_API_KEY"])
)

# Create a DNA sequence
seq = benchling.dna_sequences.create(
    DnaSequenceCreate(name="GFP-insert", bases="ATGGTGAGCAAGGGC", is_circular=False, folder_id="fld_abc123")
)
print(f"Created: {seq.name} ({seq.id})")

核心 API

1. 注册库——实体 CRUD

注册库实体包括 DNA 序列、RNA 序列、AA 序列、自定义实体和混合物。所有实体类型都遵循相同的创建、读取、更新和归档模式。

from benchling_sdk.models import DnaSequenceCreate, DnaSequenceUpdate

# Create
sequence = benchling.dna_sequences.create(
    DnaSequenceCreate(
        name="My Plasmid",
        bases="ATCGATCG",
        is_circular=True,
        folder_id="fld_abc123",
        schema_id="ts_abc123",
        fields=benchling.models.fields({"gene_name": "GFP"})
    )
)
print(f"Created: {sequence.id}")

# Read
seq = benchling.dna_sequences.get_by_id(sequence.id)
print(f"Name: {seq.name}, Length: {len(seq.bases)} bp")

# Update (partial — unspecified fields unchanged)
updated = benchling.dna_sequences.update(
    sequence_id=sequence.id,
    dna_sequence=DnaSequenceUpdate(
        name="Updated Plasmid",
        fields=benchling.models.fields({"gene_name": "mCherry"})
    )
)

# Archive
benchling.dna_sequences.archive(ids=[sequence.id], reason="RETIRED")

# Register entity in registry (with auto-generated ID)
registered = benchling.dna_sequences.create(
    DnaSequenceCreate(
        name="Production Plasmid",
        bases="ATCGATCG",
        is_circular=True,
        folder_id="fld_abc123",
        entity_registry_id="src_abc123",
        naming_strategy="NEW_IDS"  # or "IDS_FROM_NAMES"
    )
)
print(f"Registry ID: {registered.entity_registry_id}")

# Entity types available via SDK:
# benchling.dna_sequences, benchling.rna_sequences,
# benchling.aa_sequences, benchling.custom_entities, benchling.mixtures

2. 注册库——列表和分页

所有列表操作都会返回分页生成器，以提高内存使用效率。

# List with pagination
sequences = benchling.dna_sequences.list()
total = sequences.estimated_count()
print(f"Total sequences: {total}")

for page in sequences:
    for seq in page:
        print(f"  {seq.name} ({seq.id}): {len(seq.bases)} bp")

# Filter by schema
filtered = benchling.dna_sequences.list(schema_id="ts_abc123")
for page in filtered:
    for seq in page:
        print(f"  {seq.name}")

3. 库存管理

管理实物样品、容器、盒子和位置。

from benchling_sdk.models import ContainerCreate, BoxCreate

# Create container (sample tube)
container = benchling.containers.create(
    ContainerCreate(
        name="Sample Tube 001",
        schema_id="cont_schema_abc123",
        parent_storage_id="box_abc123",
        fields=benchling.models.fields({"concentration": "100 ng/uL"})
    )
)
print(f"Container: {container.id}, Barcode: {container.barcode}")

# Create box
box = benchling.boxes.create(
    BoxCreate(
        name="Freezer Box A1",
        schema_id="box_schema_abc123",
        parent_storage_id="loc_abc123"
    )
)

# Transfer container to new location
benchling.containers.transfer(
    container_id=container.id,
    destination_id="box_xyz789"
)
print(f"Transferred {container.name} to new box")

4. 笔记本条目（ELN）

创建和管理电子实验室笔记本条目。

from benchling_sdk.models import EntryCreate

# Create notebook entry
entry = benchling.entries.create(
    EntryCreate(
        name="Experiment 2026-02-17",
        folder_id="fld_abc123",
        schema_id="entry_schema_abc123",
        fields=benchling.models.fields({
            "objective": "Test gene expression levels",
            "protocol": "Standard qPCR"
        })
    )
)
print(f"Entry: {entry.id}")

# Link entity to entry
benchling.entry_links.create(
    entry_id=entry.id,
    entity_id="seq_xyz789"
)

5. 工作流自动化

创建和管理用于实验室流程自动化的工作流任务。

from benchling_sdk.models import WorkflowTaskCreate, WorkflowTaskUpdate

# Create workflow task
task = benchling.workflow_tasks.create(
    WorkflowTaskCreate(
        name="PCR Amplification",
        workflow_id="wf_abc123",
        assignee_id="user_abc123",
        fields=benchling.models.fields({"template": "seq_abc123"})
    )
)
print(f"Task: {task.id}, Status: {task.status}")

# Update task status
benchling.workflow_tasks.update(
    task_id=task.id,
    workflow_task=WorkflowTaskUpdate(status_id="status_complete_abc123")
)

# Wait for async operations
from benchling_sdk.helpers.tasks import wait_for_task

result = wait_for_task(
    benchling, task_id="task_abc123",
    interval_wait_seconds=2, max_wait_seconds=300
)
print(f"Async task completed: {result}")

6. 错误处理与重试

from benchling_sdk.retry import RetryStrategy
from benchling_sdk.errors import BenchlingError

# Custom retry strategy
benchling = Benchling(
    url="https://your-tenant.benchling.com",
    auth_method=ApiKeyAuth(os.environ["BENCHLING_API_KEY"]),
    retry_strategy=RetryStrategy(max_retries=3)
)
# SDK auto-retries on 429 (rate limit), 502, 503, 504

# Error handling
try:
    seq = benchling.dna_sequences.get_by_id("seq_nonexistent")
except BenchlingError as e:
    print(f"API error: {e.status_code} — {e.message}")

核心概念

实体类型映射

Benchling 类型	SDK 访问器	使用场景
DNA 序列	benchling.dna_sequences	质粒、引物、基因插入片段
RNA 序列	benchling.rna_sequences	mRNA、gRNA、siRNA
AA 序列	benchling.aa_sequences	蛋白质、抗体、酶
自定义实体	benchling.custom_entities	细胞系、试剂、样品
混合物	benchling.mixtures	缓冲液、培养基、化合物配方
容器	benchling.containers	试管、孔、样品瓶
盒子	benchling.boxes	储存盒、架子
条目	benchling.entries	实验室笔记本条目
工作流任务	benchling.workflow_tasks	流程步骤、任务分配

架构字段

Benchling 实体使用由架构定义的自定义字段。始终使用 fields() 辅助函数：

# Correct: use fields() helper
fields = benchling.models.fields({
    "concentration": "100 ng/uL",
    "date_prepared": "2026-02-17",
    "passage_number": 5
})

# Fields are typed by schema — string, number, date, entity link, dropdown

分页模式

所有 list() 调用都会返回分页生成器。切勿调用 list() 后不进行迭代：

# Correct: iterate through pages
for page in benchling.dna_sequences.list():
    for item in page:
        process(item)

# Get count without loading all data
count = benchling.dna_sequences.list().estimated_count()

常见工作流

工作流：从 FASTA 批量导入

import os, time
from Bio import SeqIO
from benchling_sdk.benchling import Benchling
from benchling_sdk.auth.api_key_auth import ApiKeyAuth
from benchling_sdk.models import DnaSequenceCreate

benchling = Benchling(
    url="https://your-tenant.benchling.com",
    auth_method=ApiKeyAuth(os.environ["BENCHLING_API_KEY"])
)

created = []
for record in SeqIO.parse("sequences.fasta", "fasta"):
    seq = benchling.dna_sequences.create(
        DnaSequenceCreate(
            name=record.id,
            bases=str(record.seq),
            is_circular=False,
            folder_id="fld_abc123",
            fields=benchling.models.fields({
                "description": record.description,
                "source": "FASTA import"
            })
        )
    )
    created.append(seq.id)
    time.sleep(0.5)  # Rate limit compliance
    print(f"Created: {record.id} -> {seq.id}")

print(f"Imported {len(created)} sequences")

工作流：库存审计报告

import os, csv
from benchling_sdk.benchling import Benchling
from benchling_sdk.auth.api_key_auth import ApiKeyAuth

benchling = Benchling(
    url="https://your-tenant.benchling.com",
    auth_method=ApiKeyAuth(os.environ["BENCHLING_API_KEY"])
)

audit = []
containers = benchling.containers.list(parent_storage_id="loc_freezer01")
for page in containers:
    for c in page:
        audit.append({
            "id": c.id,
            "name": c.name,
            "barcode": c.barcode,
            "location": c.parent_storage_id,
            "created": str(c.created_at)
        })

with open("inventory_audit.csv", "w", newline="") as f:
    writer = csv.DictWriter(f, fieldnames=audit[0].keys())
    writer.writeheader()
    writer.writerows(audit)
print(f"Audit complete: {len(audit)} containers")

工作流：自动化 QC 工作流

1. 列出待处理的工作流任务：benchling.workflow_tasks.list(workflow_id=..., status="pending")
2. 对于每个任务，通过 benchling.dna_sequences.get_by_id() 读取关联实体
3. 运行自动验证检查（序列长度、GC 含量、限制性位点）
4. 通过 benchling.workflow_tasks.update() 将任务状态更新为 "complete" 或 "failed"
5. 通过 benchling.entries.create() 将结果记录到笔记本条目中

关键参数

参数	功能/端点	默认值	选项	作用
folder_id	所有创建操作	必填	fld_...	新实体的目标文件夹
schema_id	所有创建操作	可选	ts_...、cont_...	定义自定义字段的架构
entity_registry_id	实体注册	可选	src_...	用于注册实体的注册表
naming_strategy	实体注册	—	NEW_IDS、IDS_FROM_NAMES	注册表 ID 的生成方式
parent_storage_id	容器、盒子	可选	box_...、loc_...	库存的存储位置
max_retries	RetryStrategy	5	0–10	失败时的重试次数
interval_wait_seconds	wait_for_task	2	1–60	异步任务的轮询间隔
max_wait_seconds	wait_for_task	300	10–3600	等待异步任务完成的最长时间

最佳实践

1. 始终使用环境变量存储凭据：切勿将 API 密钥硬编码。使用 os.environ["BENCHLING_API_KEY"]。

2. 使用 fields() 辅助函数处理自定义架构字段：原始字典无法正常工作——SDK 要求使用类型化的 Fields 对象。

3. 反模式——将所有实体加载到内存中：使用分页生成器模式。对于大型数据集，切勿将 list() 转换为 Python 列表。

4. 批量操作时加入速率限制延迟：处理超过 50 个实体时，在创建/更新调用之间插入 time.sleep(0.5)。

5. 生产应用使用 OAuth，脚本使用 API 密钥：API 密钥的权限范围限定于用户；OAuth 支持应用级权限和轮换。

6. 反模式——同时使用 entity_registry_id 和 naming_strategy：创建时两者互斥。二选一即可。

7. 显式处理 BenchlingError：捕获 SDK 异常并记录状态码和消息，以便调试。

常见示例

示例：按架构导出序列

import csv

export = []
for page in benchling.dna_sequences.list(schema_id="ts_target_schema"):
    for seq in page:
        export.append({
            "registry_id": seq.entity_registry_id,
            "name": seq.name,
            "length": len(seq.bases),
            "bases": seq.bases[:50] + "..." if len(seq.bases) > 50 else seq.bases
        })

with open("sequences_export.csv", "w", newline="") as f:
    writer = csv.DictWriter(f, fieldnames=export[0].keys())
    writer.writeheader()
    writer.writerows(export)
print(f"Exported {len(export)} sequences")

示例：按自定义字段查找实体

# Search entities with specific field values
# Note: SDK list() supports limited filtering; for complex queries use Data Warehouse
results = []
for page in benchling.custom_entities.list(schema_id="ts_cell_lines"):
    for entity in page:
        fields = entity.fields or {}
        if fields.get("organism", {}).get("value") == "Human":
            results.append(entity)
            print(f"Found: {entity.name} ({entity.id})")
print(f"Total human cell lines: {len(results)}")

示例：批量归档旧实体

import time
from datetime import datetime, timedelta

cutoff = datetime.now() - timedelta(days=365)
to_archive = []

for page in benchling.custom_entities.list():
    for entity in page:
        if entity.modified_at and entity.modified_at < cutoff:
            to_archive.append(entity.id)

# Archive in batches
batch_size = 50
for i in range(0, len(to_archive), batch_size):
    batch = to_archive[i:i+batch_size]
    benchling.custom_entities.archive(ids=batch, reason="RETIRED")
    print(f"Archived batch {i//batch_size + 1}: {len(batch)} entities")
    time.sleep(1)
print(f"Total archived: {len(to_archive)}")

故障排除

问题	原因	解决方案
401 Unauthorized	API 密钥无效或已过期	在 Benchling Profile Settings 中重新生成密钥；检查是否已设置环境变量
403 Forbidden	权限不足	API 密钥继承用户权限；检查用户在 Benchling 管理后台中的角色
404 Not Found	实体 ID 或租户 URL 错误	验证 ID 格式（seq_、fld_ 等）；检查租户 URL 是否匹配
429 Too Many Requests	超出速率限制	SDK 会自动重试；批量操作之间添加 time.sleep()
创建时忽略 fields	使用了原始字典而非 fields() 辅助函数	使用 benchling.models.fields({...}) 处理自定义架构字段
naming_strategy 错误	与 entity_registry_id 一起使用	两者互斥——二选一即可
分页导致内存问题	将所有项目收集到一个列表中	使用 for page in .list() 模式逐页迭代
OAuth 令牌已过期	客户端凭据未能刷新	SDK 会自动处理刷新；检查 client_id/secret 是否有效

相关技能

• biopython-molecular-biology — 在上传到 Benchling 之前进行本地序列分析（BLAST、比对）
• opentrons-protocol-api — 自动执行实验室协议，将样本送入 Benchling 库存

参考资料

• Benchling Python SDK 文档 — 官方 SDK 指南
• Benchling API 参考 — REST API 端点文档
• Benchling SDK PyPI — 软件包安装和版本管理