タンパク質構造API — RCSB・AlphaFold統合サービスを構築および展開する

このトピックを終えると

教科書で学んだ API の基礎、JSON、モジュールの分離、デプロイ を組み合わせて、複数のタンパク質構造データベース（RCSB PDB、AlphaFold）を1つの統合された REST APIでラップし、デプロイするサービスを作成できます。 FastAPI の開発から Docker イメージのビルド、クラウドへのデプロイまで、全サイクルを扱います。

この記事は 教育目的の一般的な例 です。実運用へのデプロイでは、可観測性、認証、拡張性の要素がより高度になります。

「PDBとAlphaFoldでは全く異なる応答になる」—単一サービスの落とし穴

あなたがタンパク質の3D構造可視化ツールを作るとします。ユーザーが遺伝子名を入力すると、そのタンパク質の構造を表示するツールです。

問題点：情報ソースが複数の場所に分散している。

RCSB PDB（Protein Data Bank）：実験的に決定された構造。X線、クライオ電子顕微鏡、NMR。4桁のPDB IDで検索。GraphQLとREST APIの両方を提供。
AlphaFold DB（EBI）：AIによる予測構造。UniProt IDで検索。個別のREST API。
UniProt：タンパク質の配列と注釈。UniProt IDで検索。個別のREST API。

各APIの応答スキーマは完全に異なります。

RCSBのGraphQL応答：

json

{"entry": {"struct": {"title": "..."}, "polymer_entities": [...]}}

AlphaFoldの応答：

json

[{"uniprotAccession": "P0DTC2", "pdbUrl": "..."}]

あなたのフロントエンドがこれらの応答をすべて処理する必要がある場合、コードは複雑になり、保守が難しくなります。この問題をあなたのAPIが代わりに隠蔽する必要があります。

コンピューターサイエンスの言葉で表現すると、あなたが作るものはファサードパターンです。複数の異質なシステムの上に、統一されたインターフェースを構築し、クライアントがそのインターフェースだけを知っていればよいようにします。各下位システムの詳細は、ファサードによって隠蔽されます。

ブラックボックスからコンポーネントへ

コンポーネント 1: FastAPI 基本フレームワーク

python

from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
from typing import Optional


app = FastAPI(title="Protein Structure API", version="1.0.0")


class StructureResponse(BaseModel):
    source: str
    identifier: str
    title: str
    organism: Optional[str] = None
    resolution_angstroms: Optional[float] = None
    method: Optional[str] = None
    download_urls: dict[str, str]
    viewer_url: str


@app.get("/health")
async def health():
    return {"status": "ok"}


@app.get("/structures/{identifier}", response_model=StructureResponse)
async def get_structure(identifier: str):
    # 統合ロジックをここに実装
    return {"source": "...", "identifier": identifier, ...}

FastAPI の利点:

Pydantic を使用したリクエストとレスポンスの自動検証
OpenAPI (Swagger) ドキュメントの自動生成 (/docs エンドポイント)
async/await のネイティブサポート
タイプヒントがそのままドキュメントに反映

コンポーネント 2: モジュールの分離

機能ごとにファイルを分割します。

text

protein_api/
├── main.py           # FastAPI アプリの定義
├── models.py         # Pydantic スキーマ
├── sources/
│   ├── __init__.py
│   ├── rcsb.py       # RCSB PDB クライアント
│   ├── alphafold.py  # AlphaFold DB クライアント
│   └── uniprot.py    # UniProt クライアント
├── services.py       # 統合ロジック
└── config.py         # 設定

sources/rcsb.py:

python

import httpx
from typing import Optional
from protein_api.models import StructureResponse


RCSB_REST = "https://data.rcsb.org/rest/v1"


async def fetch_rcsb(pdb_id: str) -> Optional[StructureResponse]:
    async with httpx.AsyncClient(timeout=30) as client:
        r = await client.get(f"{RCSB_REST}/core/entry/{pdb_id}")
        if r.status_code == 404:
            return None
        r.raise_for_status()
        data = r.json()
    
    return StructureResponse(
        source="rcsb",
        identifier=pdb_id.upper(),
        title=data.get("struct", {}).get("title", ""),
        organism=extract_organism(data),
        resolution_angstroms=data.get("rcsb_entry_info", {}).get("resolution_combined", [None])[0],
        method=data.get("exptl", [{}])[0].get("method"),
        download_urls={
            "pdb": f"https://files.rcsb.org/download/{pdb_id.upper()}.pdb",
            "cif": f"https://files.rcsb.org/download/{pdb_id.upper()}.cif"
        },
        viewer_url=f"https://www.rcsb.org/3d-view/{pdb_id.upper()}"
    )


def extract_organism(data: dict) -> Optional[str]:
    entities = data.get("polymer_entities", [])
    if not entities:
        return None
    sources = entities[0].get("rcsb_entity_source_organism", [])
    if not sources:
        return None
    return sources[0].get("ncbi_scientific_name")

sources/alphafold.py:

python

import httpx
from typing import Optional
from protein_api.models import StructureResponse


AF_API = "https://alphafold.ebi.ac.uk/api"


async def fetch_alphafold(uniprot_id: str) -> Optional[StructureResponse]:
    async with httpx.AsyncClient(timeout=30) as client:
        r = await client.get(f"{AF_API}/prediction/{uniprot_id}")
        if r.status_code == 404:
            return None
        r.raise_for_status()
        data = r.json()
    
    if not data:
        return None
    
    entry = data[0]
    return StructureResponse(
        source="alphafold",
        identifier=uniprot_id.upper(),
        title=entry.get("gene", ""),
        organism=entry.get("organismScientificName"),
        method="AlphaFold prediction",
        download_urls={
            "pdb": entry.get("pdbUrl", ""),
            "cif": entry.get("cifUrl", ""),
            "confidence": entry.get("paeImageUrl", "")
        },
        viewer_url=f"https://alphafold.ebi.ac.uk/entry/{uniprot_id.upper()}"
    )

コンポーネント 3: 統合サービスロジック

python

from protein_api.sources.rcsb import fetch_rcsb
from protein_api.sources.alphafold import fetch_alphafold


def is_pdb_id(identifier: str) -> bool:
    return len(identifier) == 4 and identifier[0].isdigit() and identifier[1:].isalnum()


def is_uniprot_id(identifier: str) -> bool:
    if len(identifier) < 6 or len(identifier) > 10:
        return False
    return identifier[0].isalpha()


async def get_structure_unified(identifier: str) -> StructureResponse:
    identifier = identifier.strip()
    
    if is_pdb_id(identifier):
        result = await fetch_rcsb(identifier)
        if result:
            return result
    
    if is_uniprot_id(identifier):
        result = await fetch_alphafold(identifier)
        if result:
            return result
    
    raise HTTPException(
        status_code=404,
        detail=f"No structure found for identifier: {identifier}"
    )

これで、main.py でこのサービスだけを呼び出します。

python

from fastapi import FastAPI
from protein_api.services import get_structure_unified
from protein_api.models import StructureResponse


app = FastAPI(title="Protein Structure API", version="1.0.0")


@app.get("/structures/{identifier}", response_model=StructureResponse)
async def get_structure(identifier: str):
    return await get_structure_unified(identifier)

コンポーネント 4: Docker によるデプロイ

Dockerfile:

dockerfile

FROM python:3.12-slim

WORKDIR /app

COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt

COPY protein_api ./protein_api

EXPOSE 8000
CMD ["uvicorn", "protein_api.main:app", "--host", "0.0.0.0", "--port", "8000"]

requirements.txt:

text

fastapi>=0.115
uvicorn[standard]>=0.32
httpx>=0.28
pydantic>=2.9

ビルドと実行:

bash

docker build -t protein-api:v1 .
docker run -p 8000:8000 protein-api:v1

http://localhost:8000/docs で Swagger UI を使用してインタラクティブにテストできます。

クラウドへのデプロイ: Fly.io、Railway、Render、Google Cloud Run はすべて Docker イメージを直接デプロイできます。たとえば、Fly.io:

bash

flyctl launch  # 最初のデプロイ
flyctl deploy  # その後のデプロイ

数分で https://your-app.fly.dev/structures/6VXX が稼働する API になります。

フェーディング — あなたが埋めるべき3つの空白

空白1：キャッシング層

同じIDの要求は繰り返される。Redisやインメモリキャッシュで応答速度を向上させる。

python

from functools import lru_cache
import time


# TODO 1: 簡単なTTLキャッシュクラスを作成する
class TTLCache:
    def __init__(self, ttl_seconds: int = 3600):
        self.store = {}
        self.ttl = ttl_seconds
    
    def get(self, key: str):
        # TODO: storeにあればタイムスタンプを確認
        # 期限切れであればdelした後Noneを返す
        # 有効であれば値を返す
        pass
    
    def set(self, key: str, value) -> None:
        # TODO: (タイムスタンプ, 値)のタプルとして保存する
        pass


cache = TTLCache(ttl_seconds=3600)


async def get_structure_cached(identifier: str) -> StructureResponse:
    cached = cache.get(identifier)
    if cached:
        return cached
    result = await get_structure_unified(identifier)
    cache.set(identifier, result)
    return result

ヒント: store[key] = (time.time(), value); if key in store: ts, val = store[key]; if time.time() - ts < self.ttl: return val; del store[key]。

空白2：複数のIDをまとめて照会

1つのリクエストで複数の構造を照会する。

python

from asyncio import gather


@app.post("/structures/batch")
async def get_batch(identifiers: list[str]) -> list[dict]:
    """
    identifiersそれぞれを並行して照会し、成功/失敗の結果を返す。
    """
    # TODO 1: 各identifierに対して、get_structure_unifiedをgatherで並行して呼び出す
    # TODO 2: 失敗したものは、エラー情報とともに結果に含める
    # TODO 3: 返却形式: [{"identifier": ..., "success": bool, "data": ..., "error": ...}]
    pass

ヒント:

python

async def try_fetch(ident):
    try:
        return {"identifier": ident, "success": True, "data": (await get_structure_unified(ident)).dict()}
    except HTTPException as e:
        return {"identifier": ident, "success": False, "error": str(e.detail)}

results = await gather(*[try_fetch(i) for i in identifiers])
return results

空白3：オブザーバビリティ（メトリクス）

各エンドポイントの応答時間と成功率をPrometheus形式で公開する。

python

from prometheus_client import Counter, Histogram, generate_latest
import time


request_count = Counter(
    "protein_api_requests_total",
    "Total requests",
    ["endpoint", "source", "status"]
)


request_duration = Histogram(
    "protein_api_request_duration_seconds",
    "Request duration",
    ["endpoint", "source"]
)


@app.middleware("http")
async def track_metrics(request, call_next):
    start = time.time()
    response = await call_next(request)
    duration = time.time() - start
    
    # TODO 1: request_count.labels(...).inc()
    # TODO 2: request_duration.labels(...).observe(duration)
    
    return response


@app.get("/metrics")
async def metrics():
    return Response(content=generate_latest(), media_type="text/plain")

ヒント: endpoint = request.url.path; status = str(response.status_code); request_count.labels(endpoint=endpoint, source="internal", status=status).inc()。

考察：実運用APIサービスとの違い

API Gateway: 実運用では、複数のマイクロサービスの手前にゲートウェイを配置します。Kong、Tyk、AWS API Gatewayなどがあります。認証、レート制限、ロギングはゲートウェイで一元的に処理されます。

サーキットブレーカー + リトライ: 外部APIの失敗に対する防御策です。以前に解説した「堅牢なパイプラインとリトライ」のパターンがここでも適用されます。

OpenAPI仕様をソースとして: FastAPIはコードから仕様を自動生成しますが、実運用では、仕様を先に記述し、コードがそれに従うというアプローチも採用されます。デザインファーストとコードファーストの議論です。

サービスメッシュ: Istio、Linkerdなどがあります。サービス間の通信に、自動的なmTLS、リトライ、可観測性の機能を注入します。

可観測性: 実運用では、ログ、メトリクス、トレースの3つの要素を組み合わせて監視します。OpenTelemetryが標準です。

AlphaFoldによるローカル予測: EBIデータベースに登録されていないタンパク質も、AlphaFoldをローカルGPUで実行して予測できます。ColabFoldが代替手段となります。

拡張プロジェクト

1. py3Dmol統合: レスポンスに3D構造の可視化HTMLスニペットを含める。

2. GraphQLインターフェース: RESTと並行してGraphQLもサポート。Strawberryライブラリを使用。

3. WebSocketストリーミング: 大きな構造ファイルをチャンク単位でストリーミングする。

4. 複数のクラウドへのデプロイ: Fly.io、Cloud Run、Vercel（Serverless Functions）に同じAPIをデプロイし、パフォーマンスとコストを比較する。

このセクションの構成要素

[F] API の基礎: REST の原則、ステータスコード、リソースとアクション。
[F] JSON: Pydantic スキーマ、統一されたレスポンス形式、snake_case と camelCase。
[F] モジュールの分離: sources/services 構造。関心の分離（SoC）。
[F] デプロイメント: Dockerfile、クラウドへのデプロイ（Fly.io / Cloud Run）。
[W] HTTP の基礎: httpx 非同期クライアント（完成したスクリプトを提供）。

[F] = 自分で実装 / [W] = 完成したコードとして提供。

BioPlayground

タンパク質構造API：RCSBとAlphaFoldを統合したサービスを作成・展開する。

タンパク質構造API — RCSB・AlphaFold統合サービスを構築および展開する

このトピックを終えると

「PDBとAlphaFoldでは全く異なる応答になる」—単一サービスの落とし穴

ブラックボックスからコンポーネントへ

コンポーネント 1: FastAPI 基本フレームワーク

コンポーネント 2: モジュールの分離

コンポーネント 3: 統合サービスロジック

コンポーネント 4: Docker によるデプロイ

フェーディング — あなたが埋めるべき3つの空白

空白1：キャッシング層

空白2：複数のIDをまとめて照会

空白3：オブザーバビリティ（メトリクス）

考察：実運用APIサービスとの違い

拡張プロジェクト

このセクションの構成要素

💬 質問・コメント