benchling-integration

Benchling 集成

概述

Benchling 是一个面向生命科学研发的云平台。通过 Python SDK 和 REST API，以编程方式访问注册表实体（DNA、蛋白质）、库存、电子实验记录本和工作流。

何时使用此技能

在以下情况下应使用此技能：

• 使用 Benchling 的 Python SDK 或 REST API
• 管理生物序列（DNA、RNA、蛋白质）和注册表实体
• 自动化库存操作（样本、容器、位置、转移）
• 创建或查询电子实验记录本条目
• 构建工作流自动化或 Benchling Apps
• 在 Benchling 与外部系统之间同步数据
• 查询 Benchling Data Warehouse 以进行分析
• 使用 AWS EventBridge 设置事件驱动的集成

核心功能

1. 身份验证与设置

Python SDK 安装：

# Stable release
uv pip install benchling-sdk
# or with 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
)

要点：

• 可在 Benchling 的个人资料设置中获取 API 密钥
• 安全存储凭据（使用环境变量或密码管理器）
• 所有 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="My Plasmid",
        bases="ATCGATCG",
        is_circular=True,
        folder_id="fld_abc123",
        schema_id="ts_abc123",  # optional
        fields=benchling.models.fields({"gene_name": "GFP"})
    )
)

注册表注册：

若要在创建实体时直接将其注册：

sequence = benchling.dna_sequences.create(
    DnaSequenceCreate(
        name="My Plasmid",
        bases="ATCGATCG",
        is_circular=True,
        folder_id="fld_abc123",
        entity_registry_id="src_abc123",  # Registry to register in
        naming_strategy="NEW_IDS"  # or "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="Updated Plasmid Name",
        fields=benchling.models.fields({"gene_name": "mCherry"})
    )
)

未指定的字段保持不变，因此可以进行部分更新。

列表与分页：

# List all DNA sequences (returns a generator)
sequences = benchling.dna_sequences.list()
for page in sequences:
    for seq in page:
        print(f"{seq.name} ({seq.id})")

# Check total count
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="Sample Tube 001",
        schema_id="cont_schema_abc123",
        parent_storage_id="box_abc123",  # optional
        fields=benchling.models.fields({"concentration": "100 ng/μL"})
    )
)

管理盒子：

from benchling_sdk.models import BoxCreate

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

转移物品：

# Transfer a container to a new location
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="Experiment 2025-10-20",
        folder_id="fld_abc123",
        schema_id="entry_schema_abc123",
        fields=benchling.models.fields({"objective": "Test gene expression"})
    )
)

将实体关联到条目：

# Add references to entities in an entry
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 Amplification",
        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"
    )
)

异步操作：

某些操作是异步的，会返回任务：

# Wait for task completion
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 事件，实现实时集成。

事件类型：

• 实体创建、更新、归档
• 库存转移
• 工作流任务状态变更
• 条目创建与更新
• 结果注册

集成模式：

1. 在 Benchling 设置中配置将事件路由到 AWS EventBridge
2. 创建 EventBridge 规则以筛选事件
3. 将事件路由到 Lambda 函数或其他目标
4. 处理事件并更新外部系统

使用场景：

• 将 Benchling 数据同步到外部数据库
• 在工作流完成时触发下游流程
• 在实体变更时发送通知
• 记录审计跟踪

请参阅 Benchling 的事件文档，了解事件架构和配置。

7. 数据仓库与分析

通过数据仓库使用 SQL 查询 Benchling 历史数据。

访问方式： Benchling 数据仓库提供对 Benchling 数据的 SQL 访问，用于分析和报告。使用提供的凭据，通过标准 SQL 客户端进行连接。

常见查询：

• 汇总实验结果
• 分析库存趋势
• 生成合规报告
• 导出数据以供外部分析

与分析工具集成：

• 使用 Jupyter notebook 进行交互式分析
• BI 工具（Tableau、Looker、PowerBI）
• 自定义仪表板

最佳实践

错误处理

SDK 会自动重试失败的请求：

# Automatic retry for 429, 502, 503, 504 status codes
# Up to 5 retries with exponential backoff
# Customize retry behavior if needed
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)
)

分页效率

使用生成器实现内存高效的分页：

# Generator-based iteration
for page in benchling.dna_sequences.list():
    for sequence in page:
        process(sequence)

# Check estimated count without loading all pages
total = benchling.dna_sequences.list().estimated_count()

架构字段辅助工具

使用 fields() 辅助工具处理自定义架构字段：

# Convert dict to Fields object
custom_fields = benchling.models.fields({
    "concentration": "100 ng/μL",
    "date_prepared": "2025-10-20",
    "notes": "High quality prep"
})

前向兼容性

SDK 可以妥善处理未知的枚举值和类型：

• 保留未知的枚举值
• 无法识别的多态类型会返回 UnknownType
• 支持使用更新版本的 API

安全注意事项

• 切勿将 API 密钥提交到版本控制系统
• 使用环境变量存储凭据
• 如果密钥泄露，请轮换密钥
• 为应用授予最低限度的必要权限
• 在多用户场景中使用 OAuth

资源

references/

详细参考文档，提供深入信息：

• authentication.md - 全面的身份验证指南，涵盖 OIDC、安全最佳实践和凭据管理
• sdk_reference.md - 详细的 Python SDK 参考，涵盖高级模式、示例和所有实体类型
• api_endpoints.md - REST API 端点参考，介绍如何在不使用 SDK 的情况下直接进行 HTTP 调用

根据具体集成需求按需加载这些参考资料。

scripts/

此技能目前包含示例脚本，可将其移除或替换为适用于特定 Benchling 工作流的自定义自动化脚本。

常见使用场景

1. 批量导入实体：

# Import multiple sequences from FASTA file
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. 库存审计：

# List all containers in a specific location
containers = benchling.containers.list(
    parent_storage_id="box_abc123"
)

for page in containers:
    for container in page:
        print(f"{container.name}: {container.barcode}")

3. 工作流自动化：

# Update all pending tasks for a workflow
tasks = benchling.workflow_tasks.list(
    workflow_id="wf_abc123",
    status="pending"
)

for page in tasks:
    for task in page:
        # Perform automated checks
        if auto_validate(task):
            benchling.workflow_tasks.update(
                task_id=task.id,
                workflow_task=WorkflowTaskUpdate(
                    status_id="status_complete"
                )
            )

4. 数据导出：

# Export all sequences with specific properties
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)
            })

# Save to CSV or database
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]