단백질 구조 API — RCSB·AlphaFold 통합 서비스를 만들고 배포하기

이 토픽을 마치면

교과서에서 배운 API 기초, JSON, 모듈 분리, 배포 를 엮어서, 여러 단백질 구조 DB(RCSB PDB, AlphaFold)를 하나의 통합 REST API로 감싸 배포하는 서비스를 만들 수 있습니다. FastAPI 개발에서 Docker 이미지 빌드, 클라우드 배포까지 전 사이클을 다룹니다.

이 글은 교육용 일반 예제 입니다. 실전 배포는 관측성·인증·확장성 요소가 훨씬 정교합니다.

"PDB랑 AlphaFold가 완전 다른 응답이네" — 단일 서비스의 함정

여러분이 단백질 3D 구조 시각화 도구를 만들고 있다고 합시다. 사용자가 유전자 이름 하나를 입력하면 그 단백질의 구조를 보여주려 합니다.

문제: 정보 소스가 여러 곳에 흩어져 있습니다.

RCSB PDB (Protein Data Bank): 실험적으로 결정된 구조. X-ray, cryo-EM, 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가 대신 감춰줘야 합니다.

CS의 언어로 표현하면, 여러분이 만들 것은 facade 패턴입니다. 여러 이질적 시스템 위에 하나의 통일된 인터페이스를 얹어, 클라이언트가 그 인터페이스만 알면 되도록 하는 것. 각 하부 시스템의 세부는 facade가 감춥니다.

블랙박스에서 부품으로

부품 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가 됩니다.

페이딩 — 여러분이 채워야 할 세 빈칸

빈칸 1: 캐싱 레이어

같은 identifier 요청은 반복. 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에 있으면 timestamp 확인
        # 만료됐으면 del 후 None
        # 유효하면 값 반환
        pass
    
    def set(self, key: str, value) -> None:
        # TODO: (timestamp, value) 튜플로 저장
        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: 여러 identifier 배치 조회

한 요청으로 여러 구조를 조회.

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: 실패한 것은 error 정보와 함께 결과에 포함
    # 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: 관측성 (metrics)

각 엔드포인트의 응답 시간·성공률을 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: 실전은 여러 마이크로서비스 앞에 gateway를 둡니다. Kong, Tyk, AWS API Gateway. 인증·rate limiting·로깅을 gateway에서 통합 처리.

Circuit Breaker + Retry: 외부 API 실패에 대한 방어. 앞서 다룬 robust-pipeline-retry 편의 패턴이 여기서도 적용.

OpenAPI 명세를 소스로: FastAPI는 코드에서 명세를 자동 생성하지만, 실전은 반대로 명세를 먼저 쓰고 코드가 그것을 따르는 접근도 씁니다. Design-first vs code-first 논쟁.

Service Mesh: Istio, Linkerd. 서비스 간 통신에 자동 mTLS, 재시도, 관측성 주입.

Observability: 실전은 로그·metrics·traces 3축을 함께 봅니다. OpenTelemetry가 표준.

AlphaFold 로컬 예측: EBI DB에 없는 단백질도 여러분이 직접 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 원칙, 상태 코드, 리소스 vs 액션.
[F] JSON: Pydantic 스키마, 통일 응답 포맷, snake_case vs camelCase.
[F] 모듈 분리: sources/services 구조. 관심사 분리(SoC).
[F] 배포: Dockerfile, 클라우드 배포 (Fly.io/Cloud Run).
[W] HTTP 기초: httpx 비동기 클라이언트 (완성 스크립트 제공).

[F] = 여러분이 직접 구현 / [W] = 완성 코드로 제공.