basic kg API service setup w participant and biosample

Author: sooyounga

.gitignore

@@ -0,0 +1,11 @@
+from fastapi import Header, HTTPException
+
+
+async def get_token_header(x_token: str = Header()):
+    if x_token != "test-token":
+        raise HTTPException(status_code=400, detail="X-Token header invalid")
+
+
+async def get_query_token(token: str):
+    if token != "sooyoung":
+        raise HTTPException(status_code=400, detail="No Sooyoung token provided")

__init__.py

@@ -0,0 +1,23 @@
+from fastapi import Depends, FastAPI
+
+from dependencies import get_query_token, get_token_header
+# from internal import admin
+from routers import biosamples, participants
+
+app = FastAPI(dependencies=[Depends(get_query_token)])
+
+
+app.include_router(participants.router)
+app.include_router(biosamples.router)
+# app.include_router(
+#     admin.router,
+#     prefix="/admin",
+#     tags=["admin"],
+#     dependencies=[Depends(get_token_header)],
+#     responses={418: {"description": "I'm a teapot"}},
+# )
+
+
+@app.get("/")
+async def root():
+    return {"message": "Knowledge Graph API"}

dependencies.py

@@ -0,0 +1,55 @@
+from fastapi import APIRouter, Depends, HTTPException
+from pydantic import BaseModel
+
+from dependencies import get_token_header
+
+router = APIRouter(
+    prefix="/biosamples",
+    tags=["biosamples"],
+    dependencies=[Depends(get_token_header)],
+    responses={404: {"description": "Not found"}},
+)
+
+
+class Biosample(BaseModel):
+    name: str
+
+
+fake_biosamples_db = {"first": {"name": "Sample 1"}, "second": {"name": "Sample 2"}}
+
+
+@router.get("/")
+async def read_biosamples():
+    return fake_biosamples_db
+
+
+@router.post("/")
+async def create_biosample(biosample: Biosample):
+    if biosample.name in fake_biosamples_db:
+        raise HTTPException(status_code=404, detail="Biosample already created")
+    return {"name": fake_biosamples_db[biosample.name]["name"], "biosample_id": biosample.name}
+
+
+@router.get("/{biosample_id}")
+async def read_biosample(biosample_id: str):
+    if biosample_id not in fake_biosamples_db:
+        raise HTTPException(status_code=404, detail="Biosample not found")
+    return {"name": fake_biosamples_db[biosample_id]["name"], "biosample_id": biosample_id}
+
+
+@router.put(
+    "/{biosample_id}",
+    tags=["custom"],
+    responses={403: {"description": "Operation forbidden"}},
+)
+async def update_biosample(biosample_id: str):
+    if biosample_id != "first":
+        raise HTTPException(
+            status_code=403, detail="You can only update the biosample: first"
+        )
+    return {"biosample_id": biosample_id, "name": "Sample 1"}
+
+
+@router.delete("/{biosample_id}")
+async def delete_biosample(biosample_id: str):
+    return None

internal/__init__.py

@@ -0,0 +1,38 @@
+from typing import Union
+
+from fastapi import APIRouter, Depends, HTTPException
+from pydantic import BaseModel
+
+from dependencies import get_token_header
+
+router = APIRouter(
+    prefix="/participants",
+    tags=["participants"],
+    dependencies=[Depends(get_token_header)],
+    responses={404: {"description": "Not found"}},
+)
+
+
+class Participant(BaseModel):
+    name: str
+    id: int
+
+
+@router.post("/", response_model=Participant)
+async def create_participant(participant: Participant):
+    return {"participant_name": participant.name, "participant_id": participant_id}
+
+
+@router.get("/{participant_id}")
+async def read_participant(participant_id: int, q: Union[str, None] = None):
+    return {"participant_id": participant_id, "q": q}
+
+
+@router.put("/{participant_id}")
+async def update_participant(participant_id: int, participant: Participant):
+    return {"participant_name": participant.name, "participant_id": participant_id}
+
+
+@router.delete("/{participant_id}")
+async def delete_participant(participant_id: int):
+    return None