BioPlayground

Volver a la lista

API de estructuras de proteínas: creación y despliegue de un servicio integrado de RCSB y AlphaFold.

Varias bases de datos de estructuras proteicas en una única API REST unificada. Desde el desarrollo con FastAPI y la separación de módulos, hasta la creación de imágenes Docker y el despliegue en la nube.

Avanzado
|
120min
|
Verificado (2026-07)
Estructura proteicaRCSB PDBAlphaFoldREST APIConsulta de estructurasFastAPIDespliegue de Docker
Progreso0/19 (0%)

API de estructuras de proteínas: Creación y despliegue de un servicio integrado RCSB·AlphaFold

Al finalizar este tema

Combinando los fundamentos de las API, JSON, modularización y despliegue aprendidos en los libros de texto, podrá crear y desplegar un servicio que encapsule varias bases de datos de estructuras proteicas (RCSB PDB, AlphaFold) en una única API REST integrada. Se cubrirá todo el ciclo de desarrollo con FastAPI, desde la creación de imágenes Docker hasta el despliegue en la nube.

Este artículo es un ejemplo genérico con fines educativos. El despliegue en entornos de producción requiere elementos de observabilidad, autenticación y escalabilidad mucho más sofisticados.

"PDB y AlphaFold devuelven respuestas completamente diferentes" — La trampa de un servicio único

Supongamos que está desarrollando una herramienta de visualización de estructuras 3D de proteínas. Su objetivo es mostrar la estructura de una proteína cuando el usuario introduce el nombre de un gen.

Problema: Las fuentes de información están dispersas.

  • RCSB PDB (Protein Data Bank): Estructuras determinadas experimentalmente (Rayos X, cryo-EM, RMN). Se consultan mediante un ID de PDB de 4 dígitos. Ofrece tanto GraphQL como API REST.
  • AlphaFold DB (EBI): Estructuras predichas por IA. Se consultan mediante el ID de UniProt. Posee su propia API REST.
  • UniProt: Secuencias y anotaciones de proteínas. Se consultan mediante el ID de UniProt. Posee su propia API REST.

El esquema de respuesta de cada API es completamente distinto.

Respuesta GraphQL de RCSB:

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

Respuesta de AlphaFold:

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

Si su frontend debe manejar ambas respuestas, el código se volverá complejo y difícil de mantener. Su API debe ocultar este problema por usted.

En términos de CS, lo que va a crear es un patrón facade. Consiste en colocar una interfaz unificada sobre varios sistemas heterogéneos, de modo que el cliente solo necesite conocer dicha interfaz. La facade oculta los detalles de cada subsistema.

De la caja negra a los componentes

Componente 1: Estructura básica de 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):
    # Lógica de integración aquí
    return {"source": "...", "identifier": identifier, ...}

Ventajas de FastAPI:

  • Validación automática de solicitudes y respuestas con Pydantic
  • Generación automática de documentación OpenAPI (Swagger) (endpoints /docs)
  • Soporte nativo de async/await
  • Documentación basada directamente en los hints de tipo

Parte 2: Separación de módulos

Divida los archivos por funcionalidad.

text
protein_api/
├── main.py           # Definición de la aplicación FastAPI
├── models.py         # Esquemas de Pydantic
├── sources/
│   ├── __init__.py
│   ├── rcsb.py       # Cliente de RCSB PDB
│   ├── alphafold.py  # Cliente de AlphaFold DB
│   └── uniprot.py    # Cliente de UniProt
├── services.py       # Lógica de integración
└── config.py         # Configuración

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: Lógica de servicio integrada

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

Ahora, solo se llama a este servicio desde 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)

Parte 4: Despliegue con 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

Compilación y ejecución:

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

En http://localhost:8000/docs puedes realizar pruebas interactivas mediante la interfaz de Swagger.

Despliegue en la nube: Fly.io, Railway, Render y Google Cloud Run permiten desplegar directamente una imagen de Docker. Por ejemplo, en Fly.io:

bash
flyctl launch  # Primer despliegue
flyctl deploy  # Despliegues posteriores

En unos minutos, https://your-app.fly.dev/structures/6VXX será una API activa.

Fading — Los tres espacios que deben completar

Espacio 1: Capa de caché

Las solicitudes con el mismo identificador se repiten. Mejore la velocidad de respuesta con Redis o una caché en memoria.

python
from functools import lru_cache
import time




# TODO 1: crear una clase sencilla de caché con TTL
class TTLCache:
    def __init__(self, ttl_seconds: int = 3600):
        self.store = {}
        self.ttl = ttl_seconds


    def get(self, key: str):
        # TODO: si está en store, comprobar timestamp
        # Si ha caducado, eliminarlo y devolver None
        # Si sigue vigente, devolver el valor
        pass


    def set(self, key: str, value) -> None:
        # TODO: guardar como tupla (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

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

Espacio en blanco 2: Consulta de múltiples identificadores en lote

Consulta varias estructuras en una sola solicitud.

python
from asyncio import gather




@app.post("/structures/batch")
async def get_batch(identifiers: list[str]) -> list[dict]:
    """
    Consulta cada elemento de identifiers en paralelo y devuelve los resultados de éxito o fallo.
    """
    # TODO 1: llamar a get_structure_unified en paralelo con gather para cada identifier
    # TODO 2: incluir los fallos en los resultados junto con la información de error
    # TODO 3: formato de retorno: [{"identifier": ..., "success": bool, "data": ..., "error": ...}]
    pass

Pista:

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

Espacio en blanco 3: Observabilidad (métricas)

Expone el tiempo de respuesta y la tasa de éxito de cada punto final en formato 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")

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

Reflexiones: Diferencias con un servicio API en producción

API Gateway: En producción, se coloca un gateway delante de múltiples microservicios (Kong, Tyk, AWS API Gateway). La autenticación, el rate limiting y el logging se gestionan de forma integrada en el gateway.

Circuit Breaker + Retry: Defensa ante fallos en APIs externas. El patrón de robust-pipeline-retry tratado anteriormente también se aplica aquí.

La especificación OpenAPI como fuente: FastAPI genera automáticamente la especificación a partir del código, pero en producción también se utiliza un enfoque donde primero se escribe la especificación y el código debe seguirla. El debate entre Design-first vs Code-first.

Service Mesh: Istio, Linkerd. Inyección automática de mTLS, reintentos y observabilidad en la comunicación entre servicios.

Observabilidad: En producción, se analizan los tres ejes: logs, métricas y trazas. OpenTelemetry es el estándar.

Predicción local de AlphaFold: Incluso para proteínas que no están en la base de datos EBI, puedes ejecutar AlphaFold directamente en tu GPU local para realizar la predicción. ColabFold es una alternativa.

Proyectos de expansión

1. Integración de py3Dmol: Incluye un snippet HTML de visualización de estructuras 3D en la respuesta.

2. Interfaz GraphQL: Soporte para GraphQL en paralelo con REST. Biblioteca Strawberry.

3. Streaming por WebSocket: Transmisión de archivos de estructuras grandes por fragmentos (chunks).

4. Despliegue en múltiples nubes: Desplegar la misma API en Fly.io, Cloud Run y Vercel (Serverless Functions) para comparar rendimiento y costes.

Mapa de componentes de este tutorial

  • [F] Fundamentos de API: Principios REST, códigos de estado, recursos vs acciones.
  • [F] JSON: Esquemas Pydantic, formato de respuesta unificado, snake_case vs camelCase.
  • [F] Separación de módulos: Estructura sources/services. Separación de responsabilidades (SoC).
  • [F] Despliegue: Dockerfile, despliegue en la nube (Fly.io/Cloud Run).
  • [W] Fundamentos de HTTP: Cliente asíncrono httpx (se proporciona el script completo).

[F] = Lo implementas tú / [W] = Se proporciona el código completo.

💬 Preguntas y comentarios

0 comentarios

Puedes publicar sin iniciar sesión. Los comentarios de invitados no pueden editarse ni eliminarse después.

0/2000

Cargando...