名称: benchling-integration 描述: “Benchling研发平台集成。通过API访问注册表(DNA、蛋白质)、库存、电子实验室笔记本条目、工作流,构建Benchling应用,查询数据仓库,实现实验室数据管理自动化。”
Benchling集成
概述
Benchling是一个用于生命科学研发的云平台。通过Python SDK和REST API,以编程方式访问注册表实体(DNA、蛋白质)、库存、电子实验室笔记本和工作流。
何时使用此技能
此技能应在以下情况下使用:
- 使用Benchling的Python SDK或REST API时
- 管理生物序列(DNA、RNA、蛋白质)和注册表实体时
- 自动化库存操作(样品、容器、位置、转移)时
- 创建或查询电子实验室笔记本条目时
- 构建工作流自动化或Benchling应用时
- 在Benchling和外部系统之间同步数据时
- 查询Benchling数据仓库以进行分析时
- 设置与AWS EventBridge的事件驱动集成时
核心能力
1. 身份验证与设置
Python SDK安装:
# 稳定版本
pip install benchling-sdk
# 或使用Poetry
poetry add benchling-sdk
身份验证方法:
API密钥身份验证(推荐用于脚本):
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("your_api_key")
)
OAuth客户端凭证(用于应用):
from benchling_sdk.auth.client_credentials_oauth2 import ClientCredentialsOAuth2
auth_method = ClientCredentialsOAuth2(
client_id="your_client_id",
client_secret="your_client_secret"
)
benchling = Benchling(
url="https://your-tenant.benchling.com",
auth_method=auth_method
)
关键点:
- API密钥从Benchling中的配置文件设置获取
- 安全存储凭据(使用环境变量或密码管理器)
- 所有API请求都需要HTTPS
- 身份验证权限与UI中的用户权限相同
有关详细身份验证信息,包括OIDC和安全最佳实践,请参考references/authentication.md。
2. 注册表与实体管理
注册表实体包括DNA序列、RNA序列、AA序列、自定义实体和混合物。SDK提供类型化类来创建和管理这些实体。
创建DNA序列:
from benchling_sdk.models import DnaSequenceCreate
sequence = benchling.dna_sequences.create(
DnaSequenceCreate(
name="我的质粒",
bases="ATCGATCG",
is_circular=True,
folder_id="fld_abc123",
schema_id="ts_abc123", # 可选
fields=benchling.models.fields({"gene_name": "GFP"})
)
)
注册表注册:
创建时直接注册实体:
sequence = benchling.dna_sequences.create(
DnaSequenceCreate(
name="我的质粒",
bases="ATCGATCG",
is_circular=True,
folder_id="fld_abc123",
entity_registry_id="src_abc123", # 注册到的注册表
naming_strategy="NEW_IDS" # 或 "IDS_FROM_NAMES"
)
)
重要: 使用entity_registry_id或naming_strategy之一,切勿同时使用。
更新实体:
from benchling_sdk.models import DnaSequenceUpdate
updated = benchling.dna_sequences.update(
sequence_id="seq_abc123",
dna_sequence=DnaSequenceUpdate(
name="更新后的质粒名称",
fields=benchling.models.fields({"gene_name": "mCherry"})
)
)
未指定的字段保持不变,允许部分更新。
列出和分页:
# 列出所有DNA序列(返回生成器)
sequences = benchling.dna_sequences.list()
for page in sequences:
for seq in page:
print(f"{seq.name} ({seq.id})")
# 检查总数
total = sequences.estimated_count()
关键操作:
- 创建:
benchling.<entity_type>.create() - 读取:
benchling.<entity_type>.get(id)或.list() - 更新:
benchling.<entity_type>.update(id, update_object) - 归档:
benchling.<entity_type>.archive(id)
实体类型:dna_sequences、rna_sequences、aa_sequences、custom_entities、mixtures
有关全面的SDK参考和高级模式,请参考references/sdk_reference.md。
3. 库存管理
管理Benchling库存系统中的物理样品、容器、盒子和位置。
创建容器:
from benchling_sdk.models import ContainerCreate
container = benchling.containers.create(
ContainerCreate(
name="样品管001",
schema_id="cont_schema_abc123",
parent_storage_id="box_abc123", # 可选
fields=benchling.models.fields({"concentration": "100 ng/μL"})
)
)
管理盒子:
from benchling_sdk.models import BoxCreate
box = benchling.boxes.create(
BoxCreate(
name="冷冻盒A1",
schema_id="box_schema_abc123",
parent_storage_id="loc_abc123"
)
)
转移物品:
# 将容器转移到新位置
transfer = benchling.containers.transfer(
container_id="cont_abc123",
destination_id="box_xyz789"
)
关键库存操作:
- 创建容器、盒子、位置、板
- 更新库存物品属性
- 在位置之间转移物品
- 签入/签出物品
- 批量操作以进行大量转移
4. 笔记本与文档
与电子实验室笔记本(ELN)条目、协议和模板交互。
创建笔记本条目:
from benchling_sdk.models import EntryCreate
entry = benchling.entries.create(
EntryCreate(
name="实验2025-10-20",
folder_id="fld_abc123",
schema_id="entry_schema_abc123",
fields=benchling.models.fields({"objective": "测试基因表达"})
)
)
将实体链接到条目:
# 在条目中添加对实体的引用
entry_link = benchling.entry_links.create(
entry_id="entry_abc123",
entity_id="seq_xyz789"
)
关键笔记本操作:
- 创建和更新实验室笔记本条目
- 管理条目模板
- 链接实体和结果到条目
- 导出条目以进行文档记录
5. 工作流与自动化
使用Benchling的工作流系统自动化实验室流程。
创建工作流任务:
from benchling_sdk.models import WorkflowTaskCreate
task = benchling.workflow_tasks.create(
WorkflowTaskCreate(
name="PCR扩增",
workflow_id="wf_abc123",
assignee_id="user_abc123",
fields=benchling.models.fields({"template": "seq_abc123"})
)
)
更新任务状态:
from benchling_sdk.models import WorkflowTaskUpdate
updated_task = benchling.workflow_tasks.update(
task_id="task_abc123",
workflow_task=WorkflowTaskUpdate(
status_id="status_complete_abc123"
)
)
异步操作:
一些操作是异步的并返回任务:
# 等待任务完成
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
)
关键工作流操作:
- 创建和管理工作流任务
- 更新任务状态和分配
- 异步执行批量操作
- 监控任务进度
6. 事件与集成
使用AWS EventBridge订阅Benchling事件以进行实时集成。
事件类型:
- 实体创建、更新、归档
- 库存转移
- 工作流任务状态更改
- 条目创建和更新
- 结果注册
集成模式:
- 在Benchling设置中配置事件路由到AWS EventBridge
- 创建EventBridge规则以过滤事件
- 将事件路由到Lambda函数或其他目标
- 处理事件并更新外部系统
用例:
- 将Benchling数据同步到外部数据库
- 在工作流完成时触发下游流程
- 在实体更改时发送通知
- 审计跟踪日志
请参考Benchling的事件文档以获取事件模式和配置。
7. 数据仓库与分析
通过数据仓库使用SQL查询历史Benchling数据。
访问方法: Benchling数据仓库通过SQL提供对Benchling数据的访问,用于分析和报告。使用标准SQL客户端和提供的凭据连接。
常见查询:
- 聚合实验结果
- 分析库存趋势
- 生成合规报告
- 导出数据以进行外部分析
与分析工具集成:
- Jupyter笔记本用于交互式分析
- BI工具(Tableau、Looker、PowerBI)
- 自定义仪表板
最佳实践
错误处理
SDK自动重试失败的请求:
# 自动重试429、502、503、504状态码
# 最多5次重试,指数退避
# 如果需要,自定义重试行为
from benchling_sdk.retry import RetryStrategy
benchling = Benchling(
url="https://your-tenant.benchling.com",
auth_method=ApiKeyAuth("your_api_key"),
retry_strategy=RetryStrategy(max_retries=3)
)
分页效率
使用生成器以提高内存效率的分页:
# 基于生成器的迭代
for page in benchling.dna_sequences.list():
for sequence in page:
process(sequence)
# 检查估计计数而不加载所有页面
total = benchling.dna_sequences.list().estimated_count()
模式字段助手
使用fields()助手处理自定义模式字段:
# 将字典转换为Fields对象
custom_fields = benchling.models.fields({
"concentration": "100 ng/μL",
"date_prepared": "2025-10-20",
"notes": "高质量准备"
})
向前兼容性
SDK优雅地处理未知枚举值和类型:
- 保留未知枚举值
- 无法识别的多态类型返回
UnknownType - 允许使用较新的API版本
安全考虑
- 切勿将API密钥提交到版本控制
- 使用环境变量存储凭据
- 如果被泄露,轮换密钥
- 为应用授予最小必要权限
- 多用户场景使用OAuth
资源
references/
详细参考文档,提供深入信息:
- authentication.md - 全面的身份验证指南,包括OIDC、安全最佳实践和凭据管理
- sdk_reference.md - 详细的Python SDK参考,包含高级模式、示例和所有实体类型
- api_endpoints.md - REST API端点参考,用于不使用SDK的直接HTTP调用
根据需要加载这些参考以满足特定集成需求。
scripts/
此技能当前包含示例脚本,可以删除或替换为针对特定Benchling工作流的自定义自动化脚本。
常见用例
1. 批量实体导入:
# 从FASTA文件导入多个序列
from Bio import SeqIO
for record in SeqIO.parse("sequences.fasta", "fasta"):
benchling.dna_sequences.create(
DnaSequenceCreate(
name=record.id,
bases=str(record.seq),
is_circular=False,
folder_id="fld_abc123"
)
)
2. 库存审计:
# 列出特定位置中的所有容器
containers = benchling.containers.list(
parent_storage_id="box_abc123"
)
for page in containers:
for container in page:
print(f"{container.name}: {container.barcode}")
3. 工作流自动化:
# 更新工作流的所有待处理任务
tasks = benchling.workflow_tasks.list(
workflow_id="wf_abc123",
status="pending"
)
for page in tasks:
for task in page:
# 执行自动检查
if auto_validate(task):
benchling.workflow_tasks.update(
task_id=task.id,
workflow_task=WorkflowTaskUpdate(
status_id="status_complete"
)
)
4. 数据导出:
# 导出具有特定属性的所有序列
sequences = benchling.dna_sequences.list()
export_data = []
for page in sequences:
for seq in page:
if seq.schema_id == "target_schema_id":
export_data.append({
"id": seq.id,
"name": seq.name,
"bases": seq.bases,
"length": len(seq.bases)
})
# 保存到CSV或数据库
import csv
with open("sequences.csv", "w") as f:
writer = csv.DictWriter(f, fieldnames=export_data[0].keys())
writer.writeheader()
writer.writerows(export_data)
附加资源
- 官方文档: https://docs.benchling.com
- Python SDK参考: https://benchling.com/sdk-docs/
- API参考: https://benchling.com/api/reference
- 支持: [email protected]