Merge 5d15d20 into 25321c2

NIPTool/server/web_app/api/api_v1/api.py → NIPTool/API/external/api/api_v1/api.py

@@ -1,6 +1,14 @@
 from fastapi import FastAPI
 
-from NIPTool.server.web_app.api.api_v1.endpoints import batches, index, sample, update, download, statistics, login
+from NIPTool.API.external.api.api_v1.endpoints import (
+    batches,
+    index,
+    sample,
+    update,
+    download,
+    statistics,
+    login,
+)
 
 app = FastAPI()
 

NIPTool/server/web_app/api/api_v1/endpoints/batches.py → NIPTool/API/external/api/api_v1/endpoints/batches.py

@@ -1,137 +1,187 @@
 from fastapi import APIRouter, Depends, Request
 from NIPTool.adapter.plugin import NiptAdapter
-from NIPTool.server.web_app.utils import *
-from NIPTool.server.constants import TRISOMI_TRESHOLDS
-from NIPTool.server.web_app.api.deps import get_nipt_adapter
+from NIPTool.crud import find
+from NIPTool.models.database import Batch, User
+from NIPTool.API.external.utils import *
+from NIPTool.API.external.constants import TRISOMI_TRESHOLDS
+from NIPTool.API.external.api.deps import get_nipt_adapter
 from fastapi.templating import Jinja2Templates
 
 router = APIRouter()
 templates = Jinja2Templates(directory="templates")
+CURRENT_USER = User(username="mayapapaya", email="mayabrandi@123.com", role="RW").dict()
 
 
 @router.post("/")
-def batches(request: Request, adapter: NiptAdapter = Depends(get_nipt_adapter)):#, user: User = Depends(get_current_active_user)):
+def batches(
+    request: Request, adapter: NiptAdapter = Depends(get_nipt_adapter)
+):  # , user: User = Depends(get_current_active_user)):
     """List of all batches"""
 
-    all_batches = list(adapter.batches())
-    return templates.TemplateResponse("batches.html",
-                                      context={"request": request, "batches": all_batches,
-                                               "current_user": 'mayapapaya',
-                                               "page_id": "all_batches"})
+    all_batches: List[Batch] = find.batches(adapter=adapter)
+    return templates.TemplateResponse(
+        "batches.html",
+        context={
+            "request": request,
+            "batches": all_batches,
+            "current_user": CURRENT_USER,
+            "page_id": "all_batches",
+        },
+    )
+
 
 @router.get("/")
-def batches(request: Request, adapter: NiptAdapter = Depends(get_nipt_adapter)): #, user: User = Depends(get_current_active_user)):
+def batches(
+    request: Request, adapter: NiptAdapter = Depends(get_nipt_adapter)
+):  # , user: User = Depends(get_current_active_user)):
     """List of all batches"""
 
-    all_batches = list(adapter.batches())
-    return templates.TemplateResponse("batches.html",
-                                      context={"request": request, "batches": all_batches,
-                                               "current_user": 'mayapapaya',
-                                               "page_id": "all_batches"})
+    all_batches: List[Batch] = find.batches(adapter=adapter)
+    return templates.TemplateResponse(
+        "batches.html",
+        context={
+            "request": request,
+            "batches": all_batches,
+            "current_user": CURRENT_USER,
+            "page_id": "all_batches",
+        },
+    )
 
 
 @router.get("/{batch_id}/")
 def batch(request: Request, batch_id: str, adapter: NiptAdapter = Depends(get_nipt_adapter)):
     """Batch view with table of all samples in the batch."""
-    samples = adapter.batch_samples(batch_id)
-    return templates.TemplateResponse('batch/tabs/table.html',
-                                      context={"request": request, "batch": adapter.batch(batch_id),
-                                               "sample_info": [get_sample_info(sample) for
-                                                               sample in samples],
-                                               "page_id": "batches",
-                                               "current_user": 'mayapapaya'})
+
+    samples: List[Sample] = find.batch_samples(batch_id=batch_id, adapter=adapter)
+    return templates.TemplateResponse(
+        "batch/tabs/table.html",
+        context={
+            "request": request,
+            "batch": find.batch(batch_id=batch_id, adapter=adapter),
+            "sample_info": [get_sample_info(sample) for sample in samples],
+            "page_id": "batches",
+            "current_user": CURRENT_USER,
+        },
+    )
+
+
+@router.post("/{batch_id}/")
+def batch(request: Request, batch_id: str, adapter: NiptAdapter = Depends(get_nipt_adapter)):
+    """Batch view with table of all samples in the batch."""
+    samples: List[Sample] = find.batch_samples(batch_id=batch_id, adapter=adapter)
+    return templates.TemplateResponse(
+        "batch/tabs/table.html",
+        context={
+            "request": request,
+            "batch": find.batch(batch_id=batch_id, adapter=adapter),
+            "sample_info": [get_sample_info(sample) for sample in samples],
+            "page_id": "batches",
+            "current_user": CURRENT_USER,
+        },
+    )
 
 
 @router.get("/{batch_id}/{ncv}")
 def NCV(request: Request, batch_id: str, ncv, adapter: NiptAdapter = Depends(get_nipt_adapter)):
     """Batch view with with NCV plot"""
-
+    batch: Batch = find.batch(batch_id=batch_id, adapter=adapter)
     return templates.TemplateResponse(
-        "batch/tabs/NCV.html", context=dict(
+        "batch/tabs/NCV.html",
+        context=dict(
             request=request,
             tris_thresholds=TRISOMI_TRESHOLDS,
-            batch=adapter.batch(batch_id),
+            batch=batch.dict(),
             chr=ncv,
             ncv_chrom_data={ncv: get_tris_cases(adapter, ncv, batch_id)},
             normal_data=get_tris_control_normal(adapter, ncv),
             abnormal_data=get_tris_control_abnormal(adapter, ncv, 0),
             page_id=f"batches_NCV{ncv}",
-            current_user='mayapapaya')
+            current_user=CURRENT_USER,
+        ),
     )
 
 
 @router.get("/batches/{batch_id}/fetal_fraction_XY")
-def fetal_fraction_XY(request: Request, batch_id: str, adapter: NiptAdapter = Depends(get_nipt_adapter)):
+def fetal_fraction_XY(
+    request: Request, batch_id: str, adapter: NiptAdapter = Depends(get_nipt_adapter)
+):
     """Batch view with fetal fraction (X against Y) plot"""
-    batch = adapter.batch(batch_id)
+    batch: Batch = find.batch(batch_id=batch_id, adapter=adapter)
     control = get_ff_control_normal(adapter)
     abnormal = get_ff_control_abnormal(adapter)
     return templates.TemplateResponse(
         "batch/tabs/FF_XY.html",
         context=dict(
             request=request,
-            current_user='mayapapaya',
+            current_user=CURRENT_USER,
             control=control,
             abnormal=abnormal,
             cases=get_ff_cases(adapter, batch_id),
             max_x=max(control["FFX"]) + 1,
             min_x=min(control["FFX"]) - 1,
-            batch=batch,
-            page_id="batches_FF_XY")
+            batch=batch.dict(),
+            page_id="batches_FF_XY",
+        ),
     )
 
 
 @router.get("/batches/{batch_id}/fetal_fraction")
-def fetal_fraction(request: Request, batch_id: str, adapter: NiptAdapter = Depends(get_nipt_adapter)):
+def fetal_fraction(
+    request: Request, batch_id: str, adapter: NiptAdapter = Depends(get_nipt_adapter)
+):
     """Batch view with fetal fraction plot"""
-    batch = adapter.batch(batch_id)
+    batch: Batch = find.batch(batch_id=batch_id, adapter=adapter)
     return templates.TemplateResponse(
         "batch/tabs/FF.html",
         context=dict(
             request=request,
-            current_user='mayapapaya',
+            current_user=CURRENT_USER,
             control=get_ff_control_normal(adapter),
             cases=get_ff_cases(adapter, batch_id),
-            batch=batch,
-            page_id="batches_FF")
+            batch=batch.dict(),
+            page_id="batches_FF",
+        ),
     )
 
 
 @router.get("/batches/{batch_id}/coverage")
 def coverage(request: Request, batch_id: str, adapter: NiptAdapter = Depends(get_nipt_adapter)):
     """Batch view with coverage plot"""
-    batch = adapter.batch(batch_id)
-    samples = list(adapter.batch_samples(batch_id))
+    batch: Batch = find.batch(batch_id=batch_id, adapter=adapter)
+    samples: List[Sample] = find.batch_samples(batch_id=batch_id, adapter=adapter)
     scatter_data = get_scatter_data_for_coverage_plot(samples)
     box_data = get_box_data_for_coverage_plot(samples)
     return templates.TemplateResponse(
         "batch/tabs/coverage.html",
         context=dict(
             request=request,
-            current_user='mayapapaya',
-            batch=batch,
+            current_user=CURRENT_USER,
+            batch=batch.dict(),
             x_axis=list(range(1, 23)),
             scatter_data=scatter_data,
             box_data=box_data,
-            page_id="batches_cov")
+            page_id="batches_cov",
+        ),
     )
 
 
 @router.get("/batches/{batch_id}/report/{coverage}")
-def report(request: Request, batch_id: str, coverage: str, adapter: NiptAdapter = Depends(get_nipt_adapter)):
+def report(
+    request: Request, batch_id: str, coverage: str, adapter: NiptAdapter = Depends(get_nipt_adapter)
+):
     """Report view, collecting all tables and plots from one batch."""
 
-    batch = adapter.batch(batch_id)
-    samples = list(adapter.batch_samples(batch_id))
+    batch: Batch = find.batch(batch_id=batch_id, adapter=adapter)
+    samples = list(find.batch_samples(batch_id=batch_id, adapter=adapter))
     scatter_data = get_scatter_data_for_coverage_plot(samples)
     box_data = get_box_data_for_coverage_plot(samples)
     control = get_ff_control_normal(adapter)
     return templates.TemplateResponse(
         "batch/report.html",
         context=dict(
             request=request,
-            current_user='mayapapaya',
-            batch=batch,
+            current_user=CURRENT_USER,
+            batch=batch.dict(),
             # NCV
             ncv_chrom_data={
                 "13": get_tris_cases(adapter, "13", batch_id),
@@ -153,5 +203,6 @@ def report(request: Request, batch_id: str, coverage: str, adapter: NiptAdapter
             x_axis=list(range(1, 23)),
             scatter_data=scatter_data,
             box_data=box_data,
-            page_id="batches")
+            page_id="batches",
+        ),
     )

NIPTool/server/web_app/api/api_v1/endpoints/download.py → NIPTool/API/external/api/api_v1/endpoints/download.py

@@ -1,6 +1,6 @@
 from fastapi import APIRouter, Depends, Request
 from NIPTool.adapter.plugin import NiptAdapter
-from NIPTool.server.web_app.api.deps import get_nipt_adapter, get_current_active_user
+from NIPTool.API.external.api.deps import get_nipt_adapter, get_current_active_user
 from fastapi.templating import Jinja2Templates
 from NIPTool.parse.batch import validate_file_path
 from pathlib import Path
@@ -11,7 +11,9 @@
 
 
 @router.get("/batch_download/{batch_id}/{file_id}")
-def batch_download(request: Request, batch_id: str, file_id: str, adapter: NiptAdapter = Depends(get_nipt_adapter)):
+def batch_download(
+    request: Request, batch_id: str, file_id: str, adapter: NiptAdapter = Depends(get_nipt_adapter)
+):
     """View for batch downloads"""
 
     batch = adapter.batch(batch_id)
@@ -23,11 +25,15 @@ def batch_download(request: Request, batch_id: str, file_id: str, adapter: NiptA
 
     path = Path(file_path)
 
-    return FileResponse(str(path.absolute()), media_type='application/octet-stream', filename=path.name)
+    return FileResponse(
+        str(path.absolute()), media_type="application/octet-stream", filename=path.name
+    )
 
 
 @router.get("/sample_download/{sample_id}/{file_id}")
-def sample_download(request: Request, sample_id: str, file_id: str, adapter: NiptAdapter = Depends(get_nipt_adapter)):
+def sample_download(
+    request: Request, sample_id: str, file_id: str, adapter: NiptAdapter = Depends(get_nipt_adapter)
+):
     """View for sample downloads"""
 
     sample = adapter.sample(sample_id)
@@ -37,4 +43,6 @@ def sample_download(request: Request, sample_id: str, file_id: str, adapter: Nip
         return RedirectResponse(request.url)
 
     file = Path(file_path)
-    return FileResponse(str(file.absolute()), media_type='application/octet-stream', filename=file.name)
+    return FileResponse(
+        str(file.absolute()), media_type="application/octet-stream", filename=file.name
+    )

NIPTool/server/web_app/api/api_v1/endpoints/login.py → NIPTool/API/external/api/api_v1/endpoints/login.py

@@ -5,15 +5,21 @@
 from fastapi.security import OAuth2PasswordRequestForm
 from NIPTool.models.server.login import Token, UserInDB, User
 
-from NIPTool.server.web_app.api.deps import get_current_active_user, authenticate_user, create_access_token, \
-    temp_get_config
+from NIPTool.API.external.api.deps import (
+    get_current_active_user,
+    authenticate_user,
+    create_access_token,
+    temp_get_config,
+)
 from datetime import timedelta
 
 router = APIRouter()
 
 
 @router.post("/token", response_model=Token)
-def login_for_access_token(form_data: OAuth2PasswordRequestForm = Depends(), config: dict = Depends(temp_get_config)):
+def login_for_access_token(
+    form_data: OAuth2PasswordRequestForm = Depends(), config: dict = Depends(temp_get_config)
+):
     user: UserInDB = authenticate_user(form_data.username, form_data.password)
     if not user:
         raise HTTPException(
@@ -32,12 +38,14 @@ def login_for_access_token(form_data: OAuth2PasswordRequestForm = Depends(), con
 @router.post("/login")
 def login(token: Token = Depends(login_for_access_token)):
     if token:
-        headers = {'Authorization': f"{token.get('token_type')} {token.get('access_token')}",
-                   "accept": "application/json"}
+        headers = {
+            "Authorization": f"{token.get('token_type')} {token.get('access_token')}",
+            "accept": "application/json",
+        }
         print(headers)
-    return RedirectResponse('../batches', headers=headers)
+    return RedirectResponse("../batches", headers=headers)
 
 
-#@router.post("/login")
-#def login(current_user: User = Security(get_current_active_user, scopes=["items", "me"])):
+# @router.post("/login")
+# def login(current_user: User = Security(get_current_active_user, scopes=["items", "me"])):
 #    return RedirectResponse('../batches')