refactor(plate endpoints) (#106) (patch)

# Description

Move session handling out of plate endpoints

genotype_api/api/endpoints/analyses.py

@@ -7,7 +7,7 @@
 from sqlmodel import Session, select
 from sqlmodel.sql.expression import Select, SelectOfScalar
 
-from genotype_api.database.crud import delete
+from genotype_api.database.crud.delete import delete_analysis
 from genotype_api.database.crud.create import create_analyses_sample_objects, create_analysis
 from genotype_api.database.crud.read import (
     check_analyses_objects,
@@ -58,7 +58,8 @@ def delete_analysis(
     current_user: User = Depends(get_active_user),
 ):
     """Delete analysis based on analysis_id"""
-    delete.delete_analysis(session=session, analysis_id=analysis_id)
+    analysis: Analysis = get_analysis_by_id(session=session, analysis_id=analysis_id)
+    delete_analysis(session=session, analysis=analysis)
     return JSONResponse(f"Deleted analysis: {analysis_id}", status_code=status.HTTP_200_OK)
 
 

genotype_api/api/endpoints/plates.py

@@ -3,7 +3,7 @@
 from datetime import datetime
 from io import BytesIO
 from pathlib import Path
-from typing import Literal
+from typing import Literal, Sequence
 
 from fastapi import APIRouter, Depends, File, HTTPException, Query, UploadFile, status
 from fastapi.responses import JSONResponse
@@ -12,13 +12,21 @@
 from sqlmodel.sql.expression import Select, SelectOfScalar
 
 from genotype_api.database.crud.create import create_analyses_sample_objects, create_plate
+from genotype_api.database.crud.delete import delete_analysis
 from genotype_api.database.crud.read import (
     check_analyses_objects,
     get_analyses_from_plate,
     get_plate,
     get_user_by_email,
+    get_plate_read_analysis_single,
+    get_ordered_plates,
 )
-from genotype_api.database.crud.update import refresh_sample_status
+from genotype_api.database.crud.update import (
+    refresh_sample_status,
+    refresh_plate,
+    update_plate_sign_off,
+)
+from genotype_api.database.filter_models.plate_models import PlateSignOff, PlateOrderParams
 from genotype_api.database.models import (
     Analysis,
     Plate,
@@ -72,7 +80,7 @@ def upload_plate(
     plate: Plate = create_plate(session=session, plate=plate_obj)
     for analysis in plate.analyses:
         refresh_sample_status(sample=analysis.sample, session=session)
-    session.refresh(plate)
+    refresh_plate(session=session, plate=plate)
     return plate
 
 
@@ -90,14 +98,13 @@ def sign_off_plate(
     """
 
     plate: Plate = get_plate(session=session, plate_id=plate_id)
-    db_user = get_user_by_email(session=session, email=current_user.email)
-    plate.signed_by = db_user.id
-    plate.signed_at = datetime.now()
-    plate.method_document = method_document
-    plate.method_version = method_version
-    session.commit()
-    session.refresh(plate)
-    return plate
+    plate_sign_off = PlateSignOff(
+        user_id=current_user.id,
+        signed_at=datetime.now(),
+        method_document=method_document,
+        method_version=method_version,
+    )
+    return update_plate_sign_off(session=session, plate=plate, plate_sign_off=plate_sign_off)
 
 
 @router.get(
@@ -128,8 +135,7 @@ def read_plate(
     current_user: User = Depends(get_active_user),
 ):
     """Display information about a plate."""
-
-    return PlateReadWithAnalysisDetailSingle.from_orm(get_plate(session=session, plate_id=plate_id))
+    return get_plate_read_analysis_single(session=session, plate_id=plate_id)
 
 
 @router.get(
@@ -145,13 +151,13 @@ async def read_plates(
     limit: int | None = 10,
     session: Session = Depends(get_session),
     current_user: User = Depends(get_active_user),
-):
+) -> Sequence[Plate]:
     """Display all plates"""
     sort_func = desc if sort_order == "descend" else asc
-    plates: list[Plate] = session.exec(
-        select(Plate).order_by(sort_func(order_by)).offset(skip).limit(limit)
-    ).all()
-
+    order_params = PlateOrderParams(order_by=order_by, skip=skip, limit=limit)
+    plates: Sequence[Plate] = get_ordered_plates(
+        session=session, order_params=order_params, sort_func=sort_func
+    )
     return plates
 
 
@@ -166,9 +172,8 @@ def delete_plate(
     analyses: list[Analysis] = get_analyses_from_plate(session=session, plate_id=plate_id)
     analyse_ids = [analyse.id for analyse in analyses]
     for analysis in analyses:
-        session.delete(analysis)
-    session.delete(plate)
-    session.commit()
+        delete_analysis(session=session, analysis=analysis)
+    delete_plate(session=session, plate=plate)
 
     return JSONResponse(
         f"Deleted plate: {plate_id} and analyses: {analyse_ids}",

genotype_api/database/crud/delete.py

@@ -11,19 +11,11 @@
 LOG = logging.getLogger(__name__)
 
 
-def delete_analysis(session: Session, analysis_id: int) -> Analysis:
-    db_analysis = session.get(Analysis, analysis_id)
-    session.delete(db_analysis)
+def delete_analysis(session: Session, analysis: Analysis) -> None:
+    session.delete(analysis)
     session.commit()
-    return db_analysis
 
 
-def delete_plate(session: Session, plate_id: int) -> Plate | None:
-    db_plate: Plate = session.get(Plate, plate_id)
-    if not db_plate:
-        LOG.info(f"Could not find plate {plate_id}")
-        return None
-    session.delete(db_plate)
+def delete_plate(session: Session, plate: Plate) -> None:
+    session.delete(plate)
     session.commit()
-    LOG.info("Plate deleted")
-    return db_plate

genotype_api/database/crud/read.py

@@ -1,11 +1,19 @@
 import logging
+from typing import Callable, Sequence
 
 from sqlalchemy import func
 from sqlmodel import Session, select
 from sqlmodel.sql.expression import Select, SelectOfScalar
 
 from genotype_api.constants import TYPES
-from genotype_api.database.models import Analysis, Plate, Sample, User
+from genotype_api.database.filter_models.plate_models import PlateOrderParams
+from genotype_api.database.models import (
+    Analysis,
+    Plate,
+    Sample,
+    User,
+    PlateReadWithAnalysisDetailSingle,
+)
 
 SelectOfScalar.inherit_cache = True
 Select.inherit_cache = True
@@ -45,6 +53,25 @@ def get_plate(session: Session, plate_id: int) -> Plate:
     return session.exec(statement).one()
 
 
+def get_plate_read_analysis_single(
+    session: Session, plate_id: int
+) -> PlateReadWithAnalysisDetailSingle:
+    plate: Plate = get_plate(session=session, plate_id=plate_id)
+    return PlateReadWithAnalysisDetailSingle.from_orm(plate)
+
+
+def get_ordered_plates(
+    session: Session, order_params: PlateOrderParams, sort_func: Callable
+) -> Sequence[Plate]:
+    plates: Sequence[Plate] = session.exec(
+        select(Plate)
+        .order_by(sort_func(order_params.order_by))
+        .offset(order_params.skip)
+        .limit(order_params.limit)
+    ).all()
+    return plates
+
+
 def get_incomplete_samples(statement: SelectOfScalar) -> SelectOfScalar:
     """Returning sample query statement for samples with less than two analyses."""
 

genotype_api/database/crud/update.py

@@ -1,7 +1,8 @@
 from sqlmodel import Session
 
+from genotype_api.database.filter_models.plate_models import PlateSignOff
 from genotype_api.match import check_sample
-from genotype_api.database.models import Sample
+from genotype_api.database.models import Sample, Plate
 from sqlmodel.sql.expression import Select, SelectOfScalar
 
 SelectOfScalar.inherit_cache = True
@@ -19,3 +20,17 @@ def refresh_sample_status(sample: Sample, session: Session) -> Sample:
     session.commit()
     session.refresh(sample)
     return sample
+
+
+def refresh_plate(session: Session, plate: Plate) -> None:
+    session.refresh(plate)
+
+
+def update_plate_sign_off(session: Session, plate: Plate, plate_sign_off: PlateSignOff) -> Plate:
+    plate.signed_by = plate_sign_off.user_id
+    plate.signed_at = plate_sign_off.signed_at
+    plate.method_document = plate_sign_off.method_document
+    plate.method_version = plate_sign_off.method_version
+    session.commit()
+    session.refresh(plate)
+    return plate

genotype_api/database/filter_models/plate_models.py

@@ -0,0 +1,19 @@
+"""Module to holds models used for plate queries."""
+
+from datetime import datetime
+from typing import Callable
+
+from pydantic import BaseModel
+
+
+class PlateSignOff(BaseModel):
+    user_id: int | None
+    signed_at: datetime = datetime.now()
+    method_document: str | None
+    method_version: str | None
+
+
+class PlateOrderParams(BaseModel):
+    skip: int | None
+    limit: int | None
+    order_by: str | None