Protein Structure API — Creating and Deploying an Integrated RCSB and AlphaFold Service

After Completing This Topic

You will be able to create a service that wraps and deploys multiple protein structure databases (RCSB PDB, AlphaFold) into a single, unified REST API by combining the API basics, JSON, module separation, and deployment concepts learned in this textbook. This covers the entire cycle of FastAPI development, from building Docker images to cloud deployment.

This article is an educational, general example. Real-world deployments require much more sophisticated observability, authentication, and scalability components.

"PDB and AlphaFold give completely different responses" — The Pitfalls of a Single Service

Let's say you're building a tool to visualize protein 3D structures. You want users to input a gene name, and your tool will display the structure of that protein.

The problem: Information sources are scattered.

RCSB PDB (Protein Data Bank): Experimentally determined structures. X-ray, cryo-EM, NMR. Searchable by 4-digit PDB ID. Provides both GraphQL and REST APIs.
AlphaFold DB (EBI): AI-predicted structures. Searchable by UniProt ID. Separate REST API.
UniProt: Protein sequences and annotations. Searchable by UniProt ID. Separate REST API.

Each API has a completely different response schema.

RCSB's GraphQL response:

json

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

AlphaFold's response:

json

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

If your front-end needs to handle both of these responses, your code will become complex and difficult to maintain. Your API should hide this problem.

In computer science terms, what you're building is a facade pattern. It involves placing a unified interface on top of several heterogeneous systems, so that the client only needs to know that interface. The facade hides the details of each underlying system.

From Black Box to Components

Component 1: FastAPI Basic Skeleton

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):
    # Integrate logic here
    return {"source": "...", "identifier": identifier, ...}

Advantages of FastAPI:

Automatic request/response validation with Pydantic
Automatic generation of OpenAPI (Swagger) documentation (/docs endpoint)
Native async/await support
Type hints are directly reflected in the documentation

Component 2: Modularization

Divide the code into separate files based on functionality.

text

protein_api/
├── main.py           # FastAPI app definition
├── models.py         # Pydantic schemas
├── sources/
│   ├── __init__.py
│   ├── rcsb.py       # RCSB PDB client
│   ├── alphafold.py  # AlphaFold DB client
│   └── uniprot.py    # UniProt client
├── services.py       # Integration logic
└── config.py         # Configuration

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()}"
    )

Component 3: Unified Service Logic

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}"
    )

Now, only call this service in 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)

Component 4: Docker Deployment

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

Build and run:

bash

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

You can interactively test with Swagger UI at http://localhost:8000/docs.

Cloud Deployment: Fly.io, Railway, Render, and Google Cloud Run all support deploying Docker images directly. For example, with Fly.io:

bash

flyctl launch  # Initial deployment
flyctl deploy  # Subsequent deployments

In a few minutes, you'll have a live API at https://your-app.fly.dev/structures/6VXX.

Fading — Three Blank Spaces for You to Fill

Blank 1: Caching Layer

The same identifier is requested repeatedly. Improve response speed using Redis or an in-memory cache.

python

from functools import lru_cache
import time


# TODO 1: Create a simple TTL cache class
class TTLCache:
    def __init__(self, ttl_seconds: int = 3600):
        self.store = {}
        self.ttl = ttl_seconds
    
    def get(self, key: str):
        # TODO: If the key exists in the store, check the timestamp.
        # If it has expired, delete it and return None.
        # If it is still valid, return the value.
        pass
    
    def set(self, key: str, value) -> None:
        # TODO: Store the value as a tuple of (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

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

Blank 2: Batch Query for Multiple Identifiers

Query multiple structures with a single request.

python

from asyncio import gather


@app.post("/structures/batch")
async def get_batch(identifiers: list[str]) -> list[dict]:
    """
    Queries each identifier in parallel and returns the success/failure results.
    """
    # TODO 1: Use `gather` to make parallel calls to `get_structure_unified` for each identifier.
    # TODO 2: Include failed requests in the results along with their error information.
    # TODO 3: Return format: `[{"identifier": ..., "success": bool, "data": ..., "error": ...}]`
    pass

Hint:

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

Blank 3: Observability (Metrics)

Expose the response time and success rate of each endpoint in Prometheus format.

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")

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

Reflection — Differences from a Production API Service

API Gateway: In a production environment, a gateway is placed in front of multiple microservices. Examples include Kong, Tyk, and AWS API Gateway. Authentication, rate limiting, and logging are handled collectively at the gateway.

Circuit Breaker + Retry: Defense against external API failures. The patterns discussed in the "robust-pipeline-retry" section also apply here.

OpenAPI Specification as the Source: While FastAPI automatically generates the specification from the code, in production, the approach of writing the specification first and then having the code follow it is also used. This is the design-first vs. code-first debate.

Service Mesh: Istio, Linkerd. Automatically injects mTLS, retries, and observability into service-to-service communication.

Observability: In a production environment, logs, metrics, and traces are viewed together as a three-pronged approach. OpenTelemetry is the standard.

AlphaFold Local Prediction: You can directly run AlphaFold on your local GPU to predict proteins that are not in the EBI database. ColabFold is an alternative.

Extension Project

1. py3Dmol Integration: Include a 3D structure visualization HTML snippet in the response.

2. GraphQL Interface: Support GraphQL in parallel with REST. Use the Strawberry library.

3. WebSocket Streaming: Stream large structure files in chunks.

4. Multiple Cloud Deployments: Deploy the same API to Fly.io, Cloud Run, and Vercel (Serverless Functions) to compare performance and cost.

Component Map for This Tutorial

[F] API Fundamentals: REST principles, status codes, resources vs. actions.
[F] JSON: Pydantic schemas, consistent response format, snake_case vs. camelCase.
[F] Modularization: sources/services structure. Separation of concerns (SoC).
[F] Deployment: Dockerfile, cloud deployment (Fly.io/Cloud Run).
[W] HTTP Fundamentals: httpx asynchronous client (complete script provided).

[F] = You implement this yourself / [W] = Provided as complete code.

BioPlayground

Developing and deploying a protein structure API integrating RCSB and AlphaFold resources.

Protein Structure API — Creating and Deploying an Integrated RCSB and AlphaFold Service

After Completing This Topic

"PDB and AlphaFold give completely different responses" — The Pitfalls of a Single Service

From Black Box to Components

Component 1: FastAPI Basic Skeleton

Component 2: Modularization

Component 3: Unified Service Logic

Component 4: Docker Deployment

Fading — Three Blank Spaces for You to Fill

Blank 1: Caching Layer

Blank 2: Batch Query for Multiple Identifiers

Blank 3: Observability (Metrics)

Reflection — Differences from a Production API Service

Extension Project

Component Map for This Tutorial

💬 Questions & Comments