Initialize dcd_mapping api server

Co-authored-by: Ben Capodanno <[email protected]>

Dockerfile

@@ -46,6 +46,17 @@ RUN pip install -e '.[dev,tests]'
 RUN pip install -U polars-lts-cpu
 # install gene normalizer with pg dependencies. TODO: can the pg dependencies be specified in pyproject.toml?
 #RUN pip install 'gene-normalizer[pg]'
+
+# not working, needs to happen after db volume is mounted
+# ENV GENE_NORM_DB_URL=postgres://postgres:postgres@db:5432/gene_normalizer
+# RUN echo "y" | gene_norm_update_remote
+
 ENV PYTHONUNBUFFERED 1
 
 ENV PYTHONPATH "${PYTHONPATH}:/usr/src/app/src"
+
+# Tell Docker that we will listen on port 8000.
+EXPOSE 8000
+
+# At container startup, run the application using uvicorn.
+CMD ["uvicorn", "api.server_main:app", "--host", "0.0.0.0", "--port", "8000"]

docker-compose-dev.yml

@@ -34,6 +34,21 @@ services:
     volumes:
       - vrs-mapping-seqrepo-dev:/usr/local/share/seqrepo
 
+  api:
+    build:
+      context: .
+    command: bash -c "uvicorn api.server_main:app --host 0.0.0.0 --port 8000 --reload"
+    depends_on:
+      - db
+      - seqrepo
+    env_file:
+      - settings/.env.dev
+    ports:
+      - "8004:8000"
+    volumes:
+      - .:/usr/src/app
+      - vrs-mapping-seqrepo-dev:/usr/local/share/seqrepo
+
 volumes:
   vrs-mapping-data-dev:
   vrs-mapping-seqrepo-dev:

pyproject.toml

@@ -42,7 +42,10 @@ dependencies = [
     "pydantic>=2",
     "python-dotenv",
     "setuptools>=68.0",  # tmp -- ensure 3.12 compatibility
-    "mavehgvs==0.6.1"
+    "mavehgvs==0.6.1",
+    "fastapi",
+    "starlette",
+    "uvicorn"
 ]
 dynamic = ["version"]
 

src/api/routers/map.py

@@ -0,0 +1,77 @@
+from cool_seq_tool.schemas import AnnotationLayer
+from fastapi import APIRouter
+
+from dcd_mapping.align import align
+from dcd_mapping.annotate import (
+    _get_computed_reference_sequence,
+    _get_mapped_reference_sequence,
+    annotate,
+)
+from dcd_mapping.mavedb_data import (
+    get_raw_scoreset_metadata,
+    get_scoreset_metadata,
+    get_scoreset_records,
+)
+from dcd_mapping.schemas import ScoreAnnotation, ScoresetMapping
+from dcd_mapping.transcripts import select_transcript
+from dcd_mapping.vrs_map import vrs_map
+
+router = APIRouter(prefix="/api/v1", tags=["mappings"], responses={404: {"description": "Not found"}})
+
+@router.post(path="/map/{urn}", status_code=200, response_model=ScoresetMapping)
+async def map_scoreset(urn: str) -> ScoresetMapping:
+    metadata = get_scoreset_metadata(urn)
+    records = get_scoreset_records(urn, True)
+
+    alignment_result = align(metadata, True)
+
+    transcript = await select_transcript(metadata, records, alignment_result)
+
+    vrs_results = vrs_map(metadata, alignment_result, records, transcript, True)
+
+    # TODO raise server error if vrs_results is None
+    if vrs_results is None:
+        return None
+
+    vrs_results = annotate(vrs_results, transcript, metadata)
+
+    raw_metadata = get_raw_scoreset_metadata(urn)
+    preferred_layers = {mapping.annotation_layer for mapping in vrs_results}
+
+    reference_sequences = {
+        layer: {"computed_reference_sequence": None, "mapped_reference_sequence": None}
+        for layer in AnnotationLayer
+    }
+
+    for layer in preferred_layers:
+        reference_sequences[layer][
+            "computed_reference_sequence"
+        ] = _get_computed_reference_sequence(urn, layer, transcript)
+        reference_sequences[layer][
+            "mapped_reference_sequence"
+        ] = _get_mapped_reference_sequence(layer, transcript, alignment_result)
+
+    mapped_scores: list[ScoreAnnotation] = []
+    for m in vrs_results:
+        if m.annotation_layer in preferred_layers:
+            # drop annotation layer from mapping object
+            mapped_scores.append(ScoreAnnotation(**m.model_dump()))
+
+    output = ScoresetMapping(
+        metadata=raw_metadata,
+        computed_protein_reference_sequence=reference_sequences[
+            AnnotationLayer.PROTEIN
+        ]["computed_reference_sequence"],
+        mapped_protein_reference_sequence=reference_sequences[AnnotationLayer.PROTEIN][
+            "mapped_reference_sequence"
+        ],
+        computed_genomic_reference_sequence=reference_sequences[
+            AnnotationLayer.GENOMIC
+        ]["computed_reference_sequence"],
+        mapped_genomic_reference_sequence=reference_sequences[AnnotationLayer.GENOMIC][
+            "mapped_reference_sequence"
+        ],
+        mapped_scores=mapped_scores,
+    )
+
+    return output

src/api/server_main.py

@@ -0,0 +1,15 @@
+"""FastAPI server file"""
+import uvicorn
+from fastapi import FastAPI
+
+from api.routers import map
+
+app = FastAPI()
+
+app.include_router(map.router)
+
+
+# If the application is not already being run within a uvicorn server, start uvicorn here.
+if __name__ == "__main__":
+    uvicorn.run(app, host="0.0.0.0", port=8000)
+