feat(pep_0585):

genotype_api/api/endpoints/analyses.py

@@ -1,27 +1,20 @@
 """Routes for analysis"""
 
 from pathlib import Path
-from typing import List
 
-from fastapi import APIRouter, Depends, status, Query, UploadFile, File
+from fastapi import APIRouter, Depends, File, Query, UploadFile, status
 from fastapi.responses import JSONResponse
+from sqlmodel import Session, select
+from sqlmodel.sql.expression import Select, SelectOfScalar
 
-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.create import create_analyses_sample_objects, create_analysis
+from genotype_api.database.crud.read import check_analyses_objects, get_analysis
 from genotype_api.database.crud.update import refresh_sample_status
+from genotype_api.database.models import Analysis, AnalysisRead, AnalysisReadWithGenotype, User
 from genotype_api.database.session_handler import get_session
 from genotype_api.file_parsing.files import check_file
-from genotype_api.database.models import (
-    Analysis,
-    AnalysisRead,
-    User,
-    AnalysisReadWithGenotype,
-)
-from sqlmodel import Session, select
-
-from genotype_api.security import get_active_user
 from genotype_api.file_parsing.vcf import SequenceAnalysis
-from sqlmodel.sql.expression import Select, SelectOfScalar
+from genotype_api.security import get_active_user
 
 SelectOfScalar.inherit_cache = True
 Select.inherit_cache = True
@@ -40,15 +33,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
 
@@ -67,7 +60,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),
@@ -78,7 +71,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,34 +3,35 @@
 from datetime import datetime
 from io import BytesIO
 from pathlib import Path
-from typing import List, Optional, Literal
+from typing import Literal, Optional
 
-from fastapi import APIRouter, Depends, File, HTTPException, UploadFile, Query, status
-from sqlalchemy import desc, asc
+from fastapi import APIRouter, Depends, File, HTTPException, Query, UploadFile, status
+from fastapi.responses import JSONResponse
+from sqlalchemy import asc, desc
 from sqlmodel import Session, select
+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.read import (
+    check_analyses_objects,
     get_analyses_from_plate,
     get_plate,
     get_user_by_email,
-    check_analyses_objects,
 )
 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.session_handler import get_session
-from genotype_api.file_parsing.excel import GenotypeAnalysis
-from genotype_api.file_parsing.files import check_file
 from genotype_api.database.models import (
     Analysis,
-    User,
     Plate,
     PlateCreate,
     PlateReadWithAnalyses,
     PlateReadWithAnalysisDetail,
     PlateReadWithAnalysisDetailSingle,
+    User,
 )
+from genotype_api.database.session_handler import get_session
+from genotype_api.file_parsing.excel import GenotypeAnalysis
+from genotype_api.file_parsing.files import check_file
 from genotype_api.security import get_active_user
-from sqlmodel.sql.expression import Select, SelectOfScalar
 
 SelectOfScalar.inherit_cache = True
 Select.inherit_cache = True
@@ -43,9 +44,6 @@ def get_plate_id_from_file(file_name: Path) -> str:
     return file_name.name.split("_", 1)[0]
 
 
-from fastapi.responses import JSONResponse
-
-
 @router.post("/plate", response_model=PlateReadWithAnalyses)
 def upload_plate(
     file: UploadFile = File(...),
@@ -66,7 +64,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)
@@ -136,7 +134,7 @@ def read_plate(
 
 @router.get(
     "/",
-    response_model=List[PlateReadWithAnalysisDetail],
+    response_model=list[PlateReadWithAnalysisDetail],
     response_model_exclude={"analyses"},
     response_model_by_alias=False,
 )
@@ -150,7 +148,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()
 
@@ -165,7 +163,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,38 +1,35 @@
-from typing import List, Optional, Literal
+from collections import Counter
+from datetime import date, timedelta
+from typing import Literal, Optional
+
 from fastapi import APIRouter, Depends, Query
 from fastapi.responses import JSONResponse
-from datetime import timedelta, date
+from sqlmodel import Session, select
+from sqlmodel.sql.expression import Select, SelectOfScalar
 from starlette import status
 
 import genotype_api.database.crud.create
 from genotype_api.constants import SEXES
-from genotype_api.database.session_handler import get_session
-from genotype_api.match import check_sample
-from genotype_api.models import (
-    SampleDetail,
-    MatchResult,
-    MatchCounts,
+from genotype_api.database.crud.read import (
+    get_commented_samples,
+    get_incomplete_samples,
+    get_plate_samples,
+    get_sample,
+    get_samples,
+    get_status_missing_samples,
 )
+from genotype_api.database.crud.update import refresh_sample_status
 from genotype_api.database.models import (
     Analysis,
     Sample,
     SampleRead,
-    User,
     SampleReadWithAnalysisDeep,
+    User,
     compare_genotypes,
 )
-from collections import Counter
-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_status_missing_samples,
-    get_sample,
-    get_samples,
-)
-from sqlmodel import Session, select
-from sqlmodel.sql.expression import Select, SelectOfScalar
+from genotype_api.database.session_handler import get_session
+from genotype_api.match import check_sample
+from genotype_api.models import MatchCounts, MatchResult, SampleDetail
 from genotype_api.security import get_active_user
 
 SelectOfScalar.inherit_cache = True
@@ -71,7 +68,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}},
@@ -95,7 +92,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:
@@ -182,7 +179,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"],
@@ -191,7 +188,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,38 +1,36 @@
 """Routes for the snps"""
 
-from genotype_api.database.models import SNP, User
-from typing import List
-
 from fastapi import APIRouter, Depends, HTTPException, Query, UploadFile
-from genotype_api.database.session_handler import get_session
 from sqlmodel import Session, delete, select
+from sqlmodel.sql.expression import Select, SelectOfScalar
 
+from genotype_api.database.models import SNP, User
+from genotype_api.database.session_handler import get_session
 from genotype_api.security import get_active_user
-from sqlmodel.sql.expression import Select, SelectOfScalar
 
 SelectOfScalar.inherit_cache = True
 Select.inherit_cache = True
 
 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,19 +1,16 @@
 """Routes for users"""
 
-from typing import List
-
 from fastapi import APIRouter, Depends, HTTPException, Query
 from pydantic import EmailStr
+from sqlmodel import Session, select
+from sqlmodel.sql.expression import Select, SelectOfScalar
 from starlette import status
 from starlette.responses import JSONResponse
 
 from genotype_api.database.crud.read import get_user
+from genotype_api.database.models import User, UserCreate, UserRead, UserReadWithPlates
 from genotype_api.database.session_handler import get_session
-from genotype_api.database.models import User, UserRead, UserCreate, UserReadWithPlates
-from sqlmodel import Session, select
-
 from genotype_api.security import get_active_user
-from sqlmodel.sql.expression import Select, SelectOfScalar
 
 SelectOfScalar.inherit_cache = True
 Select.inherit_cache = True
@@ -67,14 +64,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 +81,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)

genotype_api/database/crud/create.py

@@ -1,19 +1,11 @@
 import logging
-from typing import List
 
 from fastapi import HTTPException
 from sqlmodel import Session
-
-from genotype_api.database.models import (
-    Analysis,
-    Sample,
-    User,
-    UserCreate,
-    Plate,
-    PlateCreate,
-)
 from sqlmodel.sql.expression import Select, SelectOfScalar
 
+from genotype_api.database.models import Analysis, Plate, PlateCreate, Sample, User, UserCreate
+
 SelectOfScalar.inherit_cache = True
 Select.inherit_cache = True
 
@@ -49,7 +41,7 @@ def create_sample(session: Session, sample: Sample) -> Sample:
     return sample
 
 
-def create_analyses_sample_objects(session: Session, analyses: List[Analysis]) -> List[Sample]:
+def create_analyses_sample_objects(session: Session, analyses: list[Analysis]) -> list[Sample]:
     """creating samples in an analysis if not already in db."""
     return [
         create_sample(session=session, sample=Sample(id=analysis_obj.sample_id))