Spaces:
Sleeping
Sleeping
File size: 3,497 Bytes
2e818da | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 | from fastapi import APIRouter, HTTPException
from fastapi.responses import FileResponse
from app.schemas.composition import CheckpointUpdate, CompositionSpec, MolecularPanelSpec, RuntimeCheckpoint
from app.schemas.visual_lesson import LessonPayload, NucleicAcidSpec, ProteinLessonSpec
from app.services.visual_lesson_service import VisualLessonService
from app.services.visual_lesson_store import CheckpointConflictError, LessonNotFoundError, LessonVersionError
router = APIRouter(prefix="/api/visual-lessons", tags=["visual-lessons"])
_service: VisualLessonService | None = None
def get_visual_lesson_service() -> VisualLessonService:
global _service
if _service is None:
_service = VisualLessonService()
return _service
@router.get("/{project_id}/{lesson_id}", response_model=LessonPayload)
def get_visual_lesson(project_id: str, lesson_id: str) -> LessonPayload:
try:
return get_visual_lesson_service().load_payload(project_id, lesson_id)
except LessonNotFoundError as exc:
raise HTTPException(status_code=404, detail=str(exc)) from exc
except LessonVersionError as exc:
raise HTTPException(status_code=409, detail=str(exc)) from exc
@router.put("/{project_id}/{lesson_id}/checkpoint", response_model=RuntimeCheckpoint)
def put_visual_lesson_checkpoint(project_id: str, lesson_id: str, update: CheckpointUpdate) -> RuntimeCheckpoint:
service = get_visual_lesson_service()
try:
return service.save_checkpoint(project_id, lesson_id, update)
except LessonNotFoundError as exc:
raise HTTPException(status_code=404, detail=str(exc)) from exc
except CheckpointConflictError as exc:
raise HTTPException(status_code=409, detail=str(exc)) from exc
except (LessonVersionError, ValueError) as exc:
raise HTTPException(status_code=422, detail=str(exc)) from exc
@router.get("/{project_id}/{lesson_id}/assets/{asset_id}")
def get_visual_lesson_asset(project_id: str, lesson_id: str, asset_id: str) -> FileResponse:
service = get_visual_lesson_service()
try:
spec = service.store.load(project_id, lesson_id)
structure = spec.structure if isinstance(spec, (ProteinLessonSpec, NucleicAcidSpec)) else None
if isinstance(spec, CompositionSpec):
molecular = next((panel for panel in spec.panels if isinstance(panel, MolecularPanelSpec) and panel.asset_id == asset_id), None)
if molecular is None:
raise LessonNotFoundError("Molecular structure asset not found")
coordinate_format = molecular.coordinate_format
path = service.store.asset_path(project_id, asset_id, coordinate_format)
media_type = "application/octet-stream" if coordinate_format == "bcif" else "chemical/x-mmcif"
return FileResponse(path, media_type=media_type, filename=f"{asset_id}.{coordinate_format}")
if structure is None or structure.coordinate_asset_id != asset_id:
raise LessonNotFoundError("Molecular structure asset not found")
coordinate_format = structure.coordinate_format
path = service.store.asset_path(project_id, asset_id, coordinate_format)
media_type = "application/octet-stream" if coordinate_format == "bcif" else "chemical/x-mmcif"
return FileResponse(path, media_type=media_type, filename=f"{asset_id}.{coordinate_format}")
except LessonNotFoundError as exc:
raise HTTPException(status_code=404, detail=str(exc)) from exc
|