diff --git a/genotype_api/api/endpoints/analyses.py b/genotype_api/api/endpoints/analyses.py index c5a754f..9f49a05 100644 --- a/genotype_api/api/endpoints/analyses.py +++ b/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: diff --git a/genotype_api/api/endpoints/plates.py b/genotype_api/api/endpoints/plates.py index 016c63c..ed50762 100644 --- a/genotype_api/api/endpoints/plates.py +++ b/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) diff --git a/genotype_api/api/endpoints/samples.py b/genotype_api/api/endpoints/samples.py index 89c71e7..7f9bcff 100644 --- a/genotype_api/api/endpoints/samples.py +++ b/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( diff --git a/genotype_api/api/endpoints/snps.py b/genotype_api/api/endpoints/snps.py index 157399d..2a61804 100644 --- a/genotype_api/api/endpoints/snps.py +++ b/genotype_api/api/endpoints/snps.py @@ -1,14 +1,12 @@ """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 @@ -16,23 +14,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 = [] diff --git a/genotype_api/api/endpoints/users.py b/genotype_api/api/endpoints/users.py index 75180e6..98fc292 100644 --- a/genotype_api/api/endpoints/users.py +++ b/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) diff --git a/genotype_api/database/crud/create.py b/genotype_api/database/crud/create.py index e66527d..a1da800 100644 --- a/genotype_api/database/crud/create.py +++ b/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)) diff --git a/genotype_api/database/crud/read.py b/genotype_api/database/crud/read.py index 5615957..230c461 100644 --- a/genotype_api/database/crud/read.py +++ b/genotype_api/database/crud/read.py @@ -1,12 +1,12 @@ import logging -from typing import List, Optional +from typing import Optional 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, Sample, User, Plate -from sqlmodel.sql.expression import Select, SelectOfScalar +from genotype_api.database.models import Analysis, Plate, Sample, User SelectOfScalar.inherit_cache = True Select.inherit_cache = True @@ -14,7 +14,7 @@ LOG = logging.getLogger(__name__) -def get_analyses_from_plate(plate_id: int, session: Session) -> List[Analysis]: +def get_analyses_from_plate(plate_id: int, session: Session) -> list[Analysis]: statement = select(Analysis).where(Analysis.plate_id == plate_id) return session.exec(statement).all() @@ -91,13 +91,13 @@ def get_user_by_email(session: Session, email: str) -> Optional[User]: return session.exec(statement).first() -def get_users(session: Session, skip: int = 0, limit: int = 100) -> List[User]: +def get_users(session: Session, skip: int = 0, limit: int = 100) -> list[User]: statement = select(User).offset(skip).limit(limit) return session.exec(statement).all() def check_analyses_objects( - session: Session, analyses: List[Analysis], analysis_type: TYPES + session: Session, analyses: list[Analysis], analysis_type: TYPES ) -> None: """Raising 400 if any analysis in the list already exist in the database""" for analysis_obj in analyses: diff --git a/genotype_api/database/models.py b/genotype_api/database/models.py index 9edf2a1..4a6fe56 100644 --- a/genotype_api/database/models.py +++ b/genotype_api/database/models.py @@ -1,13 +1,13 @@ from collections import Counter from datetime import datetime -from typing import Optional, List, Dict, Tuple +from typing import Optional -from pydantic import constr, EmailStr, validator +from pydantic import EmailStr, constr, validator from sqlalchemy import Index -from sqlmodel import SQLModel, Field, Relationship +from sqlmodel import Field, Relationship, SQLModel -from genotype_api.constants import TYPES, SEXES, STATUS, CUTOFS -from genotype_api.models import SampleDetail, PlateStatusCounts +from genotype_api.constants import CUTOFS, SEXES, STATUS, TYPES +from genotype_api.models import PlateStatusCounts, SampleDetail class GenotypeBase(SQLModel): @@ -25,7 +25,7 @@ class Genotype(GenotypeBase, table=True): analysis: Optional["Analysis"] = Relationship(back_populates="genotypes") @property - def alleles(self) -> List[str]: + def alleles(self) -> list[str]: """Return sorted because we are not dealing with phased data.""" return sorted([self.allele_1, self.allele_2]) @@ -59,10 +59,10 @@ class Analysis(AnalysisBase, table=True): id: Optional[int] = Field(default=None, primary_key=True) sample: Optional["Sample"] = Relationship(back_populates="analyses") - plate: Optional[List["Plate"]] = Relationship(back_populates="analyses") - genotypes: Optional[List["Genotype"]] = Relationship(back_populates="analysis") + plate: Optional[list["Plate"]] = Relationship(back_populates="analyses") + genotypes: Optional[list["Genotype"]] = Relationship(back_populates="analysis") - def check_no_calls(self) -> Dict[str, int]: + def check_no_calls(self) -> dict[str, int]: """Check that genotypes look ok.""" calls = ["known" if genotype.is_ok else "unknown" for genotype in self.genotypes] return Counter(calls) @@ -90,7 +90,7 @@ class Sample(SampleBase, table=True): __tablename__ = "sample" id: Optional[constr(max_length=32)] = Field(default=None, primary_key=True) - analyses: Optional[List["Analysis"]] = Relationship(back_populates="sample") + analyses: Optional[list["Analysis"]] = Relationship(back_populates="sample") @property def genotype_analysis(self) -> Optional[Analysis]: @@ -146,7 +146,7 @@ class UserBase(SQLModel): class User(UserBase, table=True): __tablename__ = "user" id: Optional[int] = Field(default=None, primary_key=True) - plates: Optional[List["Plate"]] = Relationship(back_populates="user") + plates: Optional[list["Plate"]] = Relationship(back_populates="user") class UserRead(UserBase): @@ -170,7 +170,7 @@ class Plate(PlateBase, table=True): __tablename__ = "plate" id: Optional[int] = Field(default=None, primary_key=True) user: Optional["User"] = Relationship(back_populates="plates") - analyses: Optional[List["Analysis"]] = Relationship(back_populates="plate") + analyses: Optional[list["Analysis"]] = Relationship(back_populates="plate") class PlateRead(PlateBase): @@ -179,23 +179,23 @@ class PlateRead(PlateBase): class PlateCreate(PlateBase): - analyses: Optional[List[Analysis]] = [] + analyses: Optional[list[Analysis]] = [] class UserReadWithPlates(UserRead): - plates: Optional[List[Plate]] = [] + plates: Optional[list[Plate]] = [] class SampleReadWithAnalysis(SampleRead): - analyses: Optional[List[AnalysisRead]] = [] + analyses: Optional[list[AnalysisRead]] = [] class AnalysisReadWithGenotype(AnalysisRead): - genotypes: Optional[List[Genotype]] = [] + genotypes: Optional[list[Genotype]] = [] class SampleReadWithAnalysisDeep(SampleRead): - analyses: Optional[List[AnalysisReadWithGenotype]] = [] + analyses: Optional[list[AnalysisReadWithGenotype]] = [] detail: Optional[SampleDetail] @validator("detail") @@ -224,7 +224,7 @@ class AnalysisReadWithSample(AnalysisRead): sample: Optional[SampleSlim] -def compare_genotypes(genotype_1: Genotype, genotype_2: Genotype) -> Tuple[str, str]: +def compare_genotypes(genotype_1: Genotype, genotype_2: Genotype) -> tuple[str, str]: """Compare two genotypes if they have the same alleles.""" if "0" in genotype_1.alleles or "0" in genotype_2.alleles: @@ -240,11 +240,11 @@ class AnalysisReadWithSampleDeep(AnalysisRead): class PlateReadWithAnalyses(PlateRead): - analyses: Optional[List[AnalysisReadWithSample]] = [] + analyses: Optional[list[AnalysisReadWithSample]] = [] class PlateReadWithAnalysisDetail(PlateRead): - analyses: Optional[List[AnalysisReadWithSample]] = [] + analyses: Optional[list[AnalysisReadWithSample]] = [] detail: Optional[PlateStatusCounts] @validator("detail") @@ -260,7 +260,7 @@ class Config: class PlateReadWithAnalysisDetailSingle(PlateRead): - analyses: Optional[List[AnalysisReadWithSample]] = [] + analyses: Optional[list[AnalysisReadWithSample]] = [] detail: Optional[PlateStatusCounts] @validator("detail") diff --git a/genotype_api/file_parsing/excel.py b/genotype_api/file_parsing/excel.py index a88b682..3d98d84 100644 --- a/genotype_api/file_parsing/excel.py +++ b/genotype_api/file_parsing/excel.py @@ -2,15 +2,15 @@ import logging from pathlib import Path -from typing import ByteString, Iterable, List, Optional +from typing import ByteString, Iterable, Optional import openpyxl -from genotype_api.exceptions import SexConflictError -from genotype_api.database.models import Genotype, Analysis - from openpyxl.workbook import Workbook from openpyxl.worksheet.worksheet import Worksheet +from genotype_api.database.models import Analysis, Genotype +from genotype_api.exceptions import SexConflictError + LOG = logging.getLogger(__name__) @@ -31,20 +31,20 @@ def __init__(self, excel_file: ByteString, file_name: str, include_key: Optional self.wb: Workbook = openpyxl.load_workbook(filename=excel_file) self.include_key: Optional[str] = include_key self.work_sheet: Worksheet = self.find_sheet(excel_db=self.wb) - self.header_row: List[str] = self.get_header_cols(self.work_sheet) + self.header_row: list[str] = self.get_header_cols(self.work_sheet) self.snp_start: int = GenotypeAnalysis.find_column(self.header_row, pattern="rs") self.sex_start: int = GenotypeAnalysis.find_column(self.header_row, pattern="ZF_") self.sex_cols: slice = slice(self.sex_start, self.sex_start + 3) - self.rs_numbers: List[str] = self.header_row[self.snp_start :] + self.rs_numbers: list[str] = self.header_row[self.snp_start :] @staticmethod - def get_header_cols(sheet: Worksheet) -> List[str]: + def get_header_cols(sheet: Worksheet) -> list[str]: return [col.value for col in sheet[1]] @staticmethod def find_sheet(excel_db: Workbook, sheet_nr: int = -1) -> Worksheet: """Fetch the sheet nr from the database""" - work_sheets: List[str] = excel_db.sheetnames + work_sheets: list[str] = excel_db.sheetnames sheet_name: str = work_sheets[sheet_nr] LOG.info("Using sheet named %s", sheet_name) sheet = excel_db[sheet_name] @@ -52,7 +52,7 @@ def find_sheet(excel_db: Workbook, sheet_nr: int = -1) -> Worksheet: return sheet @staticmethod - def find_column(header_row: List[str], pattern="rs") -> int: + def find_column(header_row: list[str], pattern="rs") -> int: """Find the first column in a row that matches a pattern.""" LOG.debug("Fetching column with %s", pattern) for index, column in enumerate(header_row): @@ -73,7 +73,7 @@ def parse_sample_id(sample_id: str, include_key: str = None) -> Optional[str]: def generate_analyses(self) -> Iterable[Analysis]: """Loop over the rows and create one analysis for each individual""" nr_row: int - row: List[str] + row: list[str] for nr_row, row in enumerate(self.work_sheet.iter_rows()): if nr_row == 0: continue @@ -104,7 +104,7 @@ def build_genotype(rs_id: str, row_value: str) -> Genotype: return Genotype(rsnumber=rs_id, allele_1=alleles[0], allele_2=alleles[1]) @staticmethod - def parse_sex(sex_cells: List[str]) -> str: + def parse_sex(sex_cells: list[str]) -> str: """Parse the sex prediction from a sample row.""" predictions = set() # first marker @@ -138,6 +138,7 @@ def parse_sex(sex_cells: List[str]) -> str: if __name__ == "__main__": import sys + import coloredlogs coloredlogs.install() diff --git a/genotype_api/file_parsing/vcf.py b/genotype_api/file_parsing/vcf.py index 9d08d90..0670733 100644 --- a/genotype_api/file_parsing/vcf.py +++ b/genotype_api/file_parsing/vcf.py @@ -1,10 +1,12 @@ """Functions to work with VCF files""" -from typing import Dict, Iterable, List, TextIO +from typing import Iterable, TextIO -from genotype_api.database.models import Genotype as DBGenotype, Analysis from pydantic import BaseModel +from genotype_api.database.models import Analysis +from genotype_api.database.models import Genotype as DBGenotype + class Genotype(BaseModel): sample_id: str @@ -18,7 +20,7 @@ class Variant(BaseModel): id: str ref: str alt: str - genotypes: List[Genotype] + genotypes: list[Genotype] class SequenceAnalysis: @@ -44,7 +46,7 @@ def set_header(self) -> None: return def generate_analyses(self) -> Iterable[Analysis]: - analyses: Dict[str, Analysis] = { + analyses: dict[str, Analysis] = { sample_id: Analysis(type="sequence", source=self.source, sample_id=sample_id) for sample_id in self.sample_ids } @@ -70,12 +72,12 @@ def generate_variants(self) -> Iterable[Variant]: line = line.rstrip() variant_line = line.split("\t") variant_info = dict(zip(self.header_columns, variant_line)) - gt_info: List[str] = [gt_call.split(":")[0] for gt_call in variant_line[9:]] + gt_info: list[str] = [gt_call.split(":")[0] for gt_call in variant_line[9:]] variant_info["genotypes"] = self.variant_genotypes(gt_info=gt_info) yield Variant(**variant_info) - def variant_genotypes(self, gt_info: List[str]) -> Iterable[dict]: + def variant_genotypes(self, gt_info: list[str]) -> Iterable[dict]: """Build Genotype objects from vcf information.""" for sample_id, bases in zip(self.sample_ids, gt_info): bases = bases.replace("|", "/") diff --git a/genotype_api/models.py b/genotype_api/models.py index 44bbc10..5289063 100644 --- a/genotype_api/models.py +++ b/genotype_api/models.py @@ -1,4 +1,4 @@ -from typing import List, Optional +from typing import Optional from pydantic import BaseModel, validator from sqlmodel import Field @@ -35,7 +35,7 @@ class SampleDetail(BaseModel): matches: Optional[int] mismatches: Optional[int] unknown: Optional[int] - failed_snps: Optional[List[str]] + failed_snps: Optional[list[str]] stats: Optional[SampleDetailStats] status: Optional[SampleDetailStatus]