Refactor to pep 0585

genotype_api/api/endpoints/analyses.py

@@ -1,23 +1,21 @@
 """Routes for analysis"""
 
 from pathlib import Path
-from typing import List
 
 from fastapi import APIRouter, Depends, status, Query, UploadFile, File
 from fastapi.responses import JSONResponse
 
-from genotype_api.crud.analyses import (
-    get_analysis,
-    check_analyses_objects,
-    create_analysis,
-)
-from genotype_api.crud.samples import (
-    create_analyses_sample_objects,
-    refresh_sample_status,
-)
+from genotype_api.database.crud.read import get_analysis, check_analyses_objects
+from genotype_api.database.crud.create import create_analysis, create_analyses_sample_objects
+from genotype_api.database.crud.update import refresh_sample_status
 from genotype_api.database import get_session
 from genotype_api.file_parsing.files import check_file
-from genotype_api.models import Analysis, AnalysisRead, AnalysisReadWithGenotype, User
+from genotype_api.database.models.models import (
+    Analysis,
+    AnalysisRead,
+    User,
+    AnalysisReadWithGenotype,
+)
 from sqlmodel import Session, select
 
 from genotype_api.security import get_active_user
@@ -41,15 +39,15 @@ def read_analysis(
     return get_analysis(session=session, analysis_id=analysis_id)
 
 
-@router.get("/", response_model=List[AnalysisRead])
+@router.get("/", response_model=list[AnalysisRead])
 def read_analyses(
     skip: int = 0,
     limit: int = Query(default=100, lte=100),
     session: Session = Depends(get_session),
     current_user: User = Depends(get_active_user),
-) -> List[Analysis]:
+) -> list[Analysis]:
     """Return all analyses."""
-    analyses: List[Analysis] = session.exec(select(Analysis).offset(skip).limit(limit)).all()
+    analyses: list[Analysis] = session.exec(select(Analysis).offset(skip).limit(limit)).all()
 
     return analyses
 
@@ -68,7 +66,7 @@ def delete_analysis(
     return JSONResponse(f"Deleted analysis: {analysis_id}", status_code=status.HTTP_200_OK)
 
 
-@router.post("/sequence", response_model=List[Analysis])
+@router.post("/sequence", response_model=list[Analysis])
 def upload_sequence_analysis(
     file: UploadFile = File(...),
     session: Session = Depends(get_session),
@@ -79,7 +77,7 @@ def upload_sequence_analysis(
     file_name: Path = check_file(file_path=file.filename, extension=".vcf")
     content = file.file.read().decode("utf-8")
     sequence_analysis = SequenceAnalysis(vcf_file=content, source=str(file_name))
-    analyses: List[Analysis] = list(sequence_analysis.generate_analyses())
+    analyses: list[Analysis] = list(sequence_analysis.generate_analyses())
     check_analyses_objects(session=session, analyses=analyses, analysis_type="sequence")
     create_analyses_sample_objects(session=session, analyses=analyses)
     for analysis in analyses:

genotype_api/api/endpoints/plates.py

@@ -3,32 +3,29 @@
 from datetime import datetime
 from io import BytesIO
 from pathlib import Path
-from typing import List, Optional, Literal
+from typing import Optional, Literal
 
 from fastapi import APIRouter, Depends, File, HTTPException, UploadFile, Query, status
 from sqlalchemy import desc, asc
 from sqlmodel import Session, select
 
-from genotype_api.crud.analyses import (
+from genotype_api.database.crud.read import (
     get_analyses_from_plate,
+    get_plate,
+    get_user_by_email,
     check_analyses_objects,
 )
-from genotype_api.crud.samples import (
-    create_analyses_sample_objects,
-    refresh_sample_status,
-)
-from genotype_api.crud.plates import create_plate, get_plate
-from genotype_api.crud.users import get_user_by_email
+from genotype_api.database.crud.update import refresh_sample_status
+from genotype_api.database.crud.create import create_plate, create_analyses_sample_objects
 from genotype_api.database import get_session
 from genotype_api.file_parsing.excel import GenotypeAnalysis
 from genotype_api.file_parsing.files import check_file
-from genotype_api.models import (
-    Plate,
-    PlateReadWithAnalyses,
+from genotype_api.database.models.models import (
     Analysis,
-    PlateCreate,
     User,
-    PlateRead,
+    Plate,
+    PlateCreate,
+    PlateReadWithAnalyses,
     PlateReadWithAnalysisDetail,
     PlateReadWithAnalysisDetailSingle,
 )
@@ -69,7 +66,7 @@ def upload_plate(
         file_name=str(file_name),
         include_key="-CG-",
     )
-    analyses: List[Analysis] = list(excel_parser.generate_analyses())
+    analyses: list[Analysis] = list(excel_parser.generate_analyses())
     check_analyses_objects(session=session, analyses=analyses, analysis_type="genotype")
     create_analyses_sample_objects(session=session, analyses=analyses)
     plate_obj = PlateCreate(plate_id=plate_id)
@@ -139,7 +136,7 @@ def read_plate(
 
 @router.get(
     "/",
-    response_model=List[PlateReadWithAnalysisDetail],
+    response_model=list[PlateReadWithAnalysisDetail],
     response_model_exclude={"analyses"},
     response_model_by_alias=False,
 )
@@ -153,7 +150,7 @@ async def read_plates(
 ):
     """Display all plates"""
     sort_func = desc if sort_order == "descend" else asc
-    plates: List[Plate] = session.exec(
+    plates: list[Plate] = session.exec(
         select(Plate).order_by(sort_func(order_by)).offset(skip).limit(limit)
     ).all()
 
@@ -168,7 +165,7 @@ def delete_plate(
 ):
     """Delete plate."""
     plate = session.get(Plate, plate_id)
-    analyses: List[Analysis] = get_analyses_from_plate(session=session, plate_id=plate_id)
+    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)

genotype_api/api/endpoints/samples.py

@@ -1,33 +1,34 @@
-from typing import List, Optional, Literal
+from typing import Optional, Literal
 from fastapi import APIRouter, Depends, Query
 from fastapi.responses import JSONResponse
-from datetime import datetime, timedelta, date
+from datetime import timedelta, date
 from starlette import status
 
+import genotype_api.database.crud.create
 from genotype_api.constants import SEXES
 from genotype_api.database import get_session
 from genotype_api.match import check_sample
 from genotype_api.models import (
-    Sample,
-    SampleReadWithAnalysis,
-    SampleRead,
-    User,
     SampleDetail,
-    Analysis,
     MatchResult,
     MatchCounts,
+)
+from genotype_api.database.models.models import (
+    Analysis,
+    Sample,
+    SampleRead,
+    User,
     SampleReadWithAnalysisDeep,
     compare_genotypes,
 )
 from collections import Counter
-from genotype_api import crud
-from genotype_api.crud.samples import (
+from genotype_api.database.crud.update import refresh_sample_status
+from genotype_api.database.crud.read import (
     get_incomplete_samples,
     get_plate_samples,
     get_commented_samples,
-    get_sample,
     get_status_missing_samples,
-    refresh_sample_status,
+    get_sample,
     get_samples,
 )
 from sqlmodel import Session, select
@@ -70,7 +71,7 @@ def read_sample(
 
 @router.get(
     "/",
-    response_model=List[SampleReadWithAnalysisDeep],
+    response_model=list[SampleReadWithAnalysisDeep],
     response_model_by_alias=False,
     response_model_exclude={
         "analyses": {"__all__": {"genotypes": True, "source": True, "created_at": True}},
@@ -94,7 +95,7 @@ def read_samples(
     status_missing: Optional[bool] = False,
     session: Session = Depends(get_session),
     current_user: User = Depends(get_active_user),
-) -> List[Sample]:
+) -> list[Sample]:
     """Returns a list of samples matching the provided filters."""
     statement: SelectOfScalar = select(Sample).distinct().join(Analysis)
     if sample_id:
@@ -118,7 +119,7 @@ def create_sample(
     session: Session = Depends(get_session),
     current_user: User = Depends(get_active_user),
 ):
-    return crud.samples.create_sample(session=session, sample=sample)
+    return genotype_api.database.database.crud.create.create_sample(session=session, sample=sample)
 
 
 @router.put("/{sample_id}/sex", response_model=SampleRead)
@@ -181,7 +182,7 @@ def set_sample_status(
     return sample
 
 
-@router.get("/{sample_id}/match", response_model=List[MatchResult])
+@router.get("/{sample_id}/match", response_model=list[MatchResult])
 def match(
     sample_id: str,
     analysis_type: Literal["genotype", "sequence"],
@@ -190,7 +191,7 @@ def match(
     date_max: Optional[date] = date.max,
     session: Session = Depends(get_session),
     current_user: User = Depends(get_active_user),
-) -> List[MatchResult]:
+) -> list[MatchResult]:
     """Match sample genotype against all other genotypes."""
 
     all_genotypes: Analysis = session.query(Analysis).filter(

genotype_api/api/endpoints/snps.py

@@ -1,7 +1,6 @@
 """Routes for the snps"""
 
-from genotype_api.models import SNP, User
-from typing import List
+from genotype_api.database.models.models import SNP, User
 
 from fastapi import APIRouter, Depends, HTTPException, Query, File, UploadFile
 from genotype_api.database import get_session
@@ -16,23 +15,23 @@
 router = APIRouter()
 
 
-@router.get("/", response_model=List[SNP])
+@router.get("/", response_model=list[SNP])
 def read_snps(
     skip: int = 0,
     limit: int = Query(default=100, lte=100),
     session: Session = Depends(get_session),
     current_user: User = Depends(get_active_user),
-) -> List[SNP]:
+) -> list[SNP]:
     return session.exec(select(SNP).offset(skip).limit(limit)).all()
 
 
-@router.post("/", response_model=List[SNP])
+@router.post("/", response_model=list[SNP])
 async def upload_snps(
     snps_file: UploadFile,
     session: Session = Depends(get_session),
     current_user: User = Depends(get_active_user),
 ):
-    db_snps: List[SNP] = session.exec(select(SNP)).all()
+    db_snps: list[SNP] = session.exec(select(SNP)).all()
     if db_snps:
         raise HTTPException(status_code=400, detail="SNPs already uploaded")
     snps = []

genotype_api/api/endpoints/users.py

@@ -1,15 +1,13 @@
 """Routes for users"""
 
-from typing import List, Optional
-
 from fastapi import APIRouter, Depends, HTTPException, Query
 from pydantic import EmailStr
 from starlette import status
 from starlette.responses import JSONResponse
 
-from genotype_api.crud.users import get_user
+from genotype_api.database.crud.read import get_user
 from genotype_api.database import get_session
-from genotype_api.models import User, UserRead, UserCreate, UserReadWithPlates
+from genotype_api.database.models.models import User, UserRead, UserCreate, UserReadWithPlates
 from sqlmodel import Session, select
 
 from genotype_api.security import get_active_user
@@ -67,14 +65,14 @@ def change_user_email(
     return user
 
 
-@router.get("/", response_model=List[UserRead])
+@router.get("/", response_model=list[UserRead])
 def read_users(
     skip: int = 0,
     limit: int = Query(default=100, lte=100),
     session: Session = Depends(get_session),
     current_user: User = Depends(get_active_user),
-) -> List[User]:
-    users: List[User] = session.exec(select(User).offset(skip).limit(limit)).all()
+) -> list[User]:
+    users: list[User] = session.exec(select(User).offset(skip).limit(limit)).all()
     return users
 
 
@@ -84,7 +82,7 @@ def create_user(
     session: Session = Depends(get_session),
     current_user: User = Depends(get_active_user),
 ):
-    user_in_db: List[User] = session.exec(select(User).where(User.email == user.email)).all()
+    user_in_db: list[User] = session.exec(select(User).where(User.email == user.email)).all()
     if user_in_db:
         raise HTTPException(status_code=409, detail="Email already registered")
     db_user = User.from_orm(user)