alterlab-benchling

Benchling 集成

概述

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

何时使用此技能

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

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

不适用的情况

场景	改用
LabArchives 笔记本、条目、附件或备份	alterlab-labarchive
在 protocols.io 上查找、撰写或发布实验方案	alterlab-protocolsio
使用 LaminDB 管理 FAIR 数据集注册表、本体整理和数据溯源	alterlab-lamindb
在 Benchling 外解析和处理序列（FASTA/GenBank 输入/输出）	alterlab-biopython
在 DNAnexus 或 LatchBio 上运行生物信息学工作流	alterlab-dnanexus / alterlab-latchbio

核心功能

1. 身份验证与设置

安装 Python SDK：

# Stable release (benchling-sdk 1.x; current 1.25.0, requires Python >= 3.9)
uv add "benchling-sdk>=1.25,<2"
# or for a throwaway script env
uv pip install benchling-sdk

SDK 遵循 SDK 级别的语义化版本控制，而非 API 版本，因此升级主版本前请先查看更新日志。

身份验证方法：

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

auth_method = ClientCredentialsOAuth2(
    client_id="your_client_id",
    client_secret=os.environ["BENCHLING_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
from benchling_sdk.helpers.serialization_helpers import fields

sequence = benchling.dna_sequences.create(
    DnaSequenceCreate(
        name="My Plasmid",
        bases="ATCGATCG",
        is_circular=True,
        folder_id="fld_abc123",
        schema_id="ts_abc123",  # optional
        fields=fields({"gene_name": {"value": "GFP"}})
    )
)

注册表登记：

若要在创建实体时直接将其登记到注册表，请设置 registry_id（要登记到哪个注册表）以及 naming_strategy（如何分配注册表 ID）。naming_strategy 必须是 NamingStrategy 枚举；使用普通字符串会导致请求序列化失败：

from benchling_sdk.models import DnaSequenceCreate, NamingStrategy

sequence = benchling.dna_sequences.create(
    DnaSequenceCreate(
        name="My Plasmid",
        bases="ATCGATCG",
        is_circular=True,
        folder_id="fld_abc123",
        registry_id="src_abc123",                 # registry to register into
        naming_strategy=NamingStrategy.NEW_IDS,   # or NamingStrategy.IDS_FROM_NAMES
    )
)

重要提示： 触发登记的是 registry_id。另有一个 entity_registry_id 字段，用于直接指定具体的注册表 ID；而且，entity_registry_id 和 naming_strategy 不能同时设置（为 ID 分配方式二选一）。

更新实体：

from benchling_sdk.models import DnaSequenceUpdate
from benchling_sdk.helpers.serialization_helpers import fields

updated = benchling.dna_sequences.update(
    dna_sequence_id="seq_abc123",
    dna_sequence=DnaSequenceUpdate(
        name="Updated Plasmid Name",
        fields=fields({"gene_name": {"value": "mCherry"}})
    )
)

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

列表与分页：

# List DNA sequences (a PageIterator that yields pages lazily)
sequences = benchling.dna_sequences.list(schema_id="ts_abc123")  # server-side filters
for page in sequences:
    for seq in page:
        print(f"{seq.name} ({seq.id})")

# Check total count
total = sequences.estimated_count()

主要操作（方法和参数名称因实体类型而异；请参阅 references/sdk_reference.md）：

• 创建：benchling.<entity_type>.create(<CreateModel>)
• 读取：benchling.<entity_type>.get_by_id(<entity>_id=...) 或 .list(...)
• 更新：benchling.<entity_type>.update(<entity>_id=..., <entity>=<UpdateModel>)
• 归档：benchling.<entity_type>.archive(<entity>_ids=[...], reason=EntityArchiveReason.MADE_IN_ERROR)（批量）

实体类型：dna_sequences、rna_sequences、aa_sequences、custom_entities、mixtures

有关完整的 SDK 参考和高级用法模式，请参阅 references/sdk_reference.md。

3. 库存管理

在 Benchling 库存系统中管理实物样本、容器、盒子和位置。

创建容器：

from benchling_sdk.models import ContainerCreate
from benchling_sdk.helpers.serialization_helpers import fields

container = benchling.containers.create(
    ContainerCreate(
        name="Sample Tube 001",
        schema_id="cont_schema_abc123",
        parent_storage_id="box_abc123",  # optional
        fields=fields({"concentration": {"value": "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"
    )
)

转移物品：

没有 containers.transfer(...) 便捷方法。使用 transfer_into_container（单个）或 transfer_into_containers（批量，返回 TaskHelper），并传入 ContainerTransfer 请求对象，在容器之间转移内容物：

from benchling_sdk.models import ContainerTransfer

benchling.containers.transfer_into_container(
    destination_container_id="cont_xyz789",
    transfer_request=ContainerTransfer(...),  # source_entity_id/source_container_id, transfer_quantity, ...
)

若要在存储中移动容器本身（而不是转移其内容物），请通过 containers.update(...) 更新其 parent_storage_id。签入/签出是批量调用：containers.checkout(ContainersCheckout(container_ids=[...], assignee_id=...)) 和 containers.checkin(ContainersCheckin(container_ids=[...]))。

4. 实验记录与文档

与电子实验记录本（ELN）条目、实验方案和模板交互。

创建实验记录本条目：

from benchling_sdk.models import EntryCreate
from benchling_sdk.helpers.serialization_helpers import fields

entry = benchling.entries.create_entry(
    EntryCreate(
        name="Experiment 2026-09-23",
        folder_id="fld_abc123",
        schema_id="entry_schema_abc123",
        fields=fields({"objective": {"value": "Test gene expression"}})
    )
)

将实体关联到条目：

v2 API 无法向现有条目写入备注文本或 @提及：EntryUpdate 仅更改元数据（name、folder_id、schema_id、fields、author_ids）。可通过以下任一方式将实体关联到条目：

• 为条目架构添加实体链接字段并设置该字段： benchling.entries.update_entry(entry_id="entry_abc123", entry=EntryUpdate(fields=fields({"plasmid": {"value": "seq_xyz789"}})))
• 使用模板创建条目，并在创建时预填其表格：EntryCreate(..., entry_template_id="TEMPLATE_ID", initial_tables=[InitialTable(template_table_id="TABLE_ID", csv_data="...")])
• 针对这些实体记录分析结果/运行记录（参见 references/api_endpoints.md）

主要实验记录操作：

• 创建和更新实验记录本条目（create_entry、update_entry、get_entry_by_id、list_entries）
• 管理条目模板（list_entry_templates、update_entry_template）
• 通过架构字段或模板表格关联实体
• 导出条目以用于文档

5. 工作流与自动化

使用 Benchling 的工作流系统自动化实验室流程。在 v2 API 中，工作流是一个工作流任务组；任务在组内创建，并包含由架构定义的字段。

创建工作流任务：

from benchling_sdk.models import WorkflowTaskCreate
from benchling_sdk.helpers.serialization_helpers import fields

task = benchling.workflow_tasks.create(
    WorkflowTaskCreate(
        workflow_task_group_id="WORKFLOW_TASK_GROUP_ID",
        assignee_id="USER_ID",            # optional
        fields=fields({"template": {"value": "seq_abc123"}})
    )
)

更新任务状态：

from benchling_sdk.models import WorkflowTaskUpdate

updated_task = benchling.workflow_tasks.update(
    workflow_task_id="WORKFLOW_TASK_ID",
    workflow_task=WorkflowTaskUpdate(
        status_id="COMPLETE_STATUS_ID"
    )
)

异步操作：

批量/长时间运行的操作会返回 TaskHelper（benchling-sdk 1.x）。调用其 wait_for_response()，阻塞至任务成功（若要获取原始任务，则调用 wait_for_completion()）：

# A bulk call returns a TaskHelper, not a plain task id
task = benchling.dna_sequences.bulk_create(...)
result = task.wait_for_response(
    interval_wait_seconds=2,
    max_wait_seconds=300,
)

1.x 中没有顶层的 wait_for_task(benchling, task_id=...) 函数；该辅助对象附属于返回的任务对象。

主要工作流操作：

• 创建和管理工作流任务
• 更新任务状态和分配
• 异步执行批量操作
• 监控任务进度

6. 事件与集成

通过 AWS EventBridge 订阅 Benchling 事件（实体创建/更新/归档、库存转移、工作流任务状态变更、条目变更、结果注册）：在 Benchling 设置中配置事件路由，创建 EventBridge 规则以筛选事件，并将事件路由到 Lambda 函数或其他目标，用于更新外部系统（同步到数据库、触发下游流程、发送通知、记录审计日志）。有关事件架构，请参阅 Benchling 的事件文档。

7. 数据仓库与分析

Benchling 数据仓库通过只读 SQL 提供对租户数据的访问，用于分析和报告（汇总结果、库存趋势、合规报告、导出）。使用按用户签发的数据仓库凭据，通过标准 PostgreSQL 客户端或 BI 工具（Jupyter、Tableau、Looker、Power BI）连接（benchling.users.get_warehouse_logins(user_id) 可列出这些凭据）。

最佳实践

错误处理

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.helpers.retry_helpers import RetryStrategy

benchling = Benchling(
    url="https://your-tenant.benchling.com",
    auth_method=ApiKeyAuth(os.environ["BENCHLING_API_KEY"]),
    retry_strategy=RetryStrategy(max_tries=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
from benchling_sdk.helpers.serialization_helpers import fields

custom_fields = fields({
    "concentration": {"value": "100 ng/μL"},
    "date_prepared": {"value": "2026-09-23"},
    "notes": {"value": "High quality prep"}
})

向前兼容

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

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

安全注意事项

• 如果密钥泄露，请轮换密钥，并且仅向应用授予其所需的权限
• 对于多用户或服务集成，优先使用 OAuth 应用，而非个人 API 密钥
• 注册表和笔记本数据可能包含未发表的知识产权或受监管的数据；导出前请遵循所在机构的数据处理规定

资源

按需加载 references/：authentication.md（OIDC、安全最佳实践、凭据管理）、sdk_reference.md（高级 SDK 模式、所有实体类型）、api_endpoints.md（用于直接 HTTP 调用的 REST 端点）。

常见用例

1. 批量导入实体：

对于大量记录，优先使用单次 bulk_create（返回 TaskHelper），而不是逐条循环调用 create——前者只需提交一个异步任务，而非发出 N 个请求。请注意，批量调用接受的是 DnaSequenceBulkCreate 对象，而不是 DnaSequenceCreate：

# Import multiple sequences from a FASTA file in one bulk job
from Bio import SeqIO
from benchling_sdk.models import DnaSequenceBulkCreate

to_create = [
    DnaSequenceBulkCreate(
        name=record.id,
        bases=str(record.seq),
        is_circular=False,
        folder_id="fld_abc123",
    )
    for record in SeqIO.parse("sequences.fasta", "fasta")
]

task = benchling.dna_sequences.bulk_create(to_create)
task.wait_for_response()  # blocks until the bulk job finishes

（对于少量记录，使用 DnaSequenceCreate，通过普通的 benchling.dna_sequences.create(...) 循环即可。）

2. 库存审计：

# List all containers stored anywhere under a box or location
containers = benchling.containers.list(
    ancestor_storage_id="box_abc123"
)

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

3. 工作流自动化：

from benchling_sdk.models import WorkflowTaskUpdate

# Update all pending tasks in a workflow task group.
# Filter by status via status_ids (a list of status IDs, not a status name)
tasks = benchling.workflow_tasks.list(
    workflow_task_group_ids=["WORKFLOW_TASK_GROUP_ID"],
    status_ids=["PENDING_STATUS_ID"]
)

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

4. 数据导出：

# Export all sequences of one schema (filtered server-side)
export_data = []
for page in benchling.dna_sequences.list(schema_id="ts_target_schema"):
    for seq in page:
        export_data.append({
            "id": seq.id,
            "name": seq.name,
            "bases": seq.bases,
            "length": seq.length,
        })

# Then write export_data to CSV/database as needed (e.g. csv.DictWriter).

其他资源

• 官方文档： https://docs.benchling.com
• Python SDK 参考： https://benchling.com/sdk-docs/
• API 参考： https://benchling.com/api/reference
• 帮助中心 / 支持： https://help.benchling.com

属于 AlterLab Academic Skills 套件。