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 — パッケージのインストールとバージョン管理