Skip to content

Commit

Permalink
feat(pep_0604): (#103)
Browse files Browse the repository at this point in the history
### Changed

- Refactor to pep0605
  • Loading branch information
henrikstranneheim authored Mar 14, 2024
1 parent c8ab511 commit 75abe98
Show file tree
Hide file tree
Showing 8 changed files with 100 additions and 108 deletions.
10 changes: 5 additions & 5 deletions genotype_api/api/endpoints/plates.py
Original file line number Diff line number Diff line change
Expand Up @@ -3,7 +3,7 @@
from datetime import datetime
from io import BytesIO
from pathlib import Path
from typing import Literal, Optional
from typing import Literal

from fastapi import APIRouter, Depends, File, HTTPException, Query, UploadFile, status
from fastapi.responses import JSONResponse
Expand Down Expand Up @@ -139,10 +139,10 @@ def read_plate(
response_model_by_alias=False,
)
async def read_plates(
order_by: Optional[Literal["created_at", "plate_id", "signed_at", "id"]] = "id",
sort_order: Optional[Literal["ascend", "descend"]] = "descend",
skip: Optional[int] = 0,
limit: Optional[int] = 10,
order_by: Literal["created_at", "plate_id", "signed_at", "id"] | None = "id",
sort_order: Literal["ascend", "descend"] | None = "descend",
skip: int | None = 0,
limit: int | None = 10,
session: Session = Depends(get_session),
current_user: User = Depends(get_active_user),
):
Expand Down
22 changes: 11 additions & 11 deletions genotype_api/api/endpoints/samples.py
Original file line number Diff line number Diff line change
@@ -1,6 +1,6 @@
from collections import Counter
from datetime import date, timedelta
from typing import Literal, Optional
from typing import Literal

from fastapi import APIRouter, Depends, Query
from fastapi.responses import JSONResponse
Expand Down Expand Up @@ -85,11 +85,11 @@ def read_sample(
def read_samples(
skip: int = 0,
limit: int = Query(default=10, lte=10),
sample_id: Optional[str] = None,
plate_id: Optional[str] = None,
incomplete: Optional[bool] = False,
commented: Optional[bool] = False,
status_missing: Optional[bool] = False,
sample_id: str | None = None,
plate_id: str | None = None,
incomplete: bool | None = False,
commented: bool | None = False,
status_missing: bool | None = False,
session: Session = Depends(get_session),
current_user: User = Depends(get_active_user),
) -> list[Sample]:
Expand Down Expand Up @@ -123,8 +123,8 @@ def create_sample(
def update_sex(
sample_id: str,
sex: SEXES = Query(...),
genotype_sex: Optional[SEXES] = None,
sequence_sex: Optional[SEXES] = None,
genotype_sex: SEXES | None = None,
sequence_sex: SEXES | None = None,
session: Session = Depends(get_session),
current_user: User = Depends(get_active_user),
):
Expand Down Expand Up @@ -166,7 +166,7 @@ def update_comment(
def set_sample_status(
sample_id: str,
session: Session = Depends(get_session),
status: Optional[Literal["pass", "fail", "cancel"]] = None,
status: Literal["pass", "fail", "cancel"] | None = None,
current_user: User = Depends(get_active_user),
):
"""Check sample analyses and update sample status accordingly."""
Expand All @@ -184,8 +184,8 @@ def match(
sample_id: str,
analysis_type: Literal["genotype", "sequence"],
comparison_set: Literal["genotype", "sequence"],
date_min: Optional[date] = date.min,
date_max: Optional[date] = date.max,
date_min: date | None = date.min,
date_max: date | None = date.max,
session: Session = Depends(get_session),
current_user: User = Depends(get_active_user),
) -> list[MatchResult]:
Expand Down
5 changes: 2 additions & 3 deletions genotype_api/database/crud/delete.py
Original file line number Diff line number Diff line change
@@ -1,10 +1,9 @@
import logging
from typing import Optional

from sqlmodel import Session
from sqlmodel.sql.expression import Select, SelectOfScalar

from genotype_api.database.models import Analysis, Plate
from sqlmodel.sql.expression import Select, SelectOfScalar

SelectOfScalar.inherit_cache = True
Select.inherit_cache = True
Expand All @@ -19,7 +18,7 @@ def delete_analysis(session: Session, analysis_id: int) -> Analysis:
return db_analysis


def delete_plate(session: Session, plate_id: int) -> Optional[Plate]:
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}")
Expand Down
5 changes: 2 additions & 3 deletions genotype_api/database/crud/read.py
Original file line number Diff line number Diff line change
@@ -1,5 +1,4 @@
import logging
from typing import Optional

from sqlalchemy import func
from sqlmodel import Session, select
Expand All @@ -21,7 +20,7 @@ def get_analyses_from_plate(plate_id: int, session: Session) -> list[Analysis]:

def get_analysis_type_sample(
sample_id: str, analysis_type: str, session: Session
) -> Optional[Analysis]:
) -> Analysis | None:
statement = select(Analysis).where(
Analysis.sample_id == sample_id, Analysis.type == analysis_type
)
Expand Down Expand Up @@ -86,7 +85,7 @@ def get_user(session: Session, user_id: int):
return session.exec(statement).one()


def get_user_by_email(session: Session, email: str) -> Optional[User]:
def get_user_by_email(session: Session, email: str) -> User | None:
statement = select(User).where(User.email == email)
return session.exec(statement).first()

Expand Down
105 changes: 52 additions & 53 deletions genotype_api/database/models.py
Original file line number Diff line number Diff line change
@@ -1,6 +1,5 @@
from collections import Counter
from datetime import datetime
from typing import Optional

from pydantic import EmailStr, constr, validator
from sqlalchemy import Index
Expand All @@ -11,18 +10,18 @@


class GenotypeBase(SQLModel):
rsnumber: Optional[constr(max_length=10)]
analysis_id: Optional[int] = Field(default=None, foreign_key="analysis.id")
allele_1: Optional[constr(max_length=1)]
allele_2: Optional[constr(max_length=1)]
rsnumber: constr(max_length=10) | None
analysis_id: int | None = Field(default=None, foreign_key="analysis.id")
allele_1: constr(max_length=1) | None
allele_2: constr(max_length=1) | None


class Genotype(GenotypeBase, table=True):
__tablename__ = "genotype"
__table_args__ = (Index("_analysis_rsnumber", "analysis_id", "rsnumber", unique=True),)
id: Optional[int] = Field(default=None, primary_key=True)
id: int | None = Field(default=None, primary_key=True)

analysis: Optional["Analysis"] = Relationship(back_populates="genotypes")
analysis: "Analysis" = Relationship(back_populates="genotypes")

@property
def alleles(self) -> list[str]:
Expand All @@ -46,21 +45,21 @@ class GenotypeCreate(GenotypeBase):

class AnalysisBase(SQLModel):
type: TYPES
source: Optional[str]
sex: Optional[SEXES]
created_at: Optional[datetime] = datetime.now()
sample_id: Optional[constr(max_length=32)] = Field(default=None, foreign_key="sample.id")
plate_id: Optional[str] = Field(default=None, foreign_key="plate.id")
source: str | None
sex: SEXES | None
created_at: datetime | None = datetime.now()
sample_id: constr(max_length=32) | None = Field(default=None, foreign_key="sample.id")
plate_id: str | None = Field(default=None, foreign_key="plate.id")


class Analysis(AnalysisBase, table=True):
__tablename__ = "analysis"
__table_args__ = (Index("_sample_type", "sample_id", "type", unique=True),)
id: Optional[int] = Field(default=None, primary_key=True)
id: int | None = 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")
sample: "Sample" = Relationship(back_populates="analyses")
plate: list["Plate"] = Relationship(back_populates="analyses")
genotypes: list["Genotype"] = Relationship(back_populates="analysis")

def check_no_calls(self) -> dict[str, int]:
"""Check that genotypes look ok."""
Expand All @@ -77,23 +76,23 @@ class AnalysisCreate(AnalysisBase):


class SampleSlim(SQLModel):
status: Optional[STATUS]
comment: Optional[str]
status: STATUS | None
comment: str | None


class SampleBase(SampleSlim):
sex: Optional[SEXES]
created_at: Optional[datetime] = datetime.now()
sex: SEXES | None
created_at: datetime | None = datetime.now()


class Sample(SampleBase, table=True):
__tablename__ = "sample"
id: Optional[constr(max_length=32)] = Field(default=None, primary_key=True)
id: constr(max_length=32) | None = Field(default=None, primary_key=True)

analyses: Optional[list["Analysis"]] = Relationship(back_populates="sample")
analyses: list["Analysis"] = Relationship(back_populates="sample")

@property
def genotype_analysis(self) -> Optional[Analysis]:
def genotype_analysis(self) -> Analysis | None:
"""Return genotype analysis."""

for analysis in self.analyses:
Expand All @@ -103,7 +102,7 @@ def genotype_analysis(self) -> Optional[Analysis]:
return None

@property
def sequence_analysis(self) -> Optional[Analysis]:
def sequence_analysis(self) -> Analysis | None:
"""Return sequence analysis."""

for analysis in self.analyses:
Expand All @@ -122,16 +121,16 @@ class SampleCreate(SampleBase):


class SNPBase(SQLModel):
ref: Optional[constr(max_length=1)]
chrom: Optional[constr(max_length=5)]
pos: Optional[int]
ref: constr(max_length=1) | None
chrom: constr(max_length=5) | None
pos: int | None


class SNP(SNPBase, table=True):
__tablename__ = "snp"
"""Represent a SNP position under investigation."""

id: Optional[constr(max_length=32)] = Field(default=None, primary_key=True)
id: constr(max_length=32) | None = Field(default=None, primary_key=True)


class SNPRead(SNPBase):
Expand All @@ -140,13 +139,13 @@ class SNPRead(SNPBase):

class UserBase(SQLModel):
email: EmailStr = Field(index=True, unique=True)
name: Optional[str] = ""
name: str | None = ""


class User(UserBase, table=True):
__tablename__ = "user"
id: Optional[int] = Field(default=None, primary_key=True)
plates: Optional[list["Plate"]] = Relationship(back_populates="user")
id: int | None = Field(default=None, primary_key=True)
plates: list["Plate"] = Relationship(back_populates="user")


class UserRead(UserBase):
Expand All @@ -158,45 +157,45 @@ class UserCreate(UserBase):


class PlateBase(SQLModel):
created_at: Optional[datetime] = datetime.now()
created_at: datetime | None = datetime.now()
plate_id: constr(max_length=16) = Field(index=True, unique=True)
signed_by: Optional[int] = Field(default=None, foreign_key="user.id")
signed_at: Optional[datetime]
method_document: Optional[str]
method_version: Optional[str]
signed_by: int | None = Field(default=None, foreign_key="user.id")
signed_at: datetime | None
method_document: str | None
method_version: str | None


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")
id: int | None = Field(default=None, primary_key=True)
user: "User" = Relationship(back_populates="plates")
analyses: list["Analysis"] = Relationship(back_populates="plate")


class PlateRead(PlateBase):
id: str
user: Optional[UserRead]
user: UserRead | None


class PlateCreate(PlateBase):
analyses: Optional[list[Analysis]] = []
analyses: list[Analysis] | None = []


class UserReadWithPlates(UserRead):
plates: Optional[list[Plate]] = []
plates: list[Plate] | None = []


class SampleReadWithAnalysis(SampleRead):
analyses: Optional[list[AnalysisRead]] = []
analyses: list[AnalysisRead] | None = []


class AnalysisReadWithGenotype(AnalysisRead):
genotypes: Optional[list[Genotype]] = []
genotypes: list[Genotype] | None = []


class SampleReadWithAnalysisDeep(SampleRead):
analyses: Optional[list[AnalysisReadWithGenotype]] = []
detail: Optional[SampleDetail]
analyses: list[AnalysisReadWithGenotype] | None = []
detail: SampleDetail | None

@validator("detail")
def get_detail(cls, value, values) -> SampleDetail:
Expand All @@ -221,7 +220,7 @@ class Config:


class AnalysisReadWithSample(AnalysisRead):
sample: Optional[SampleSlim]
sample: SampleSlim | None


def compare_genotypes(genotype_1: Genotype, genotype_2: Genotype) -> tuple[str, str]:
Expand All @@ -236,16 +235,16 @@ def compare_genotypes(genotype_1: Genotype, genotype_2: Genotype) -> tuple[str,


class AnalysisReadWithSampleDeep(AnalysisRead):
sample: Optional[SampleReadWithAnalysisDeep]
sample: SampleReadWithAnalysisDeep | None


class PlateReadWithAnalyses(PlateRead):
analyses: Optional[list[AnalysisReadWithSample]] = []
analyses: list[AnalysisReadWithSample] | None = []


class PlateReadWithAnalysisDetail(PlateRead):
analyses: Optional[list[AnalysisReadWithSample]] = []
detail: Optional[PlateStatusCounts]
analyses: list[AnalysisReadWithSample] | None = []
detail: PlateStatusCounts | None

@validator("detail")
def check_detail(cls, value, values):
Expand All @@ -260,8 +259,8 @@ class Config:


class PlateReadWithAnalysisDetailSingle(PlateRead):
analyses: Optional[list[AnalysisReadWithSample]] = []
detail: Optional[PlateStatusCounts]
analyses: list[AnalysisReadWithSample] | None = []
detail: PlateStatusCounts | None

@validator("detail")
def check_detail(cls, value, values):
Expand Down
8 changes: 4 additions & 4 deletions genotype_api/file_parsing/excel.py
Original file line number Diff line number Diff line change
Expand Up @@ -2,7 +2,7 @@

import logging
from pathlib import Path
from typing import ByteString, Iterable, Optional
from typing import ByteString, Iterable

import openpyxl
from openpyxl.workbook import Workbook
Expand All @@ -25,11 +25,11 @@ class GenotypeAnalysis:
"""

def __init__(self, excel_file: ByteString, file_name: str, include_key: Optional[str] = None):
def __init__(self, excel_file: ByteString, file_name: str, include_key: str | None = None):
LOG.info("Loading genotype information from %s", excel_file)
self.source: str = file_name
self.wb: Workbook = openpyxl.load_workbook(filename=excel_file)
self.include_key: Optional[str] = include_key
self.include_key: str | None = 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.snp_start: int = GenotypeAnalysis.find_column(self.header_row, pattern="rs")
Expand Down Expand Up @@ -60,7 +60,7 @@ def find_column(header_row: list[str], pattern="rs") -> int:
return index

@staticmethod
def parse_sample_id(sample_id: str, include_key: str = None) -> Optional[str]:
def parse_sample_id(sample_id: str, include_key: str = None) -> str | None:
"""Build samples from Excel sheet."""
LOG.info("Parse sample id from %s using include_key: %s", sample_id, include_key)
if include_key:
Expand Down
Loading

0 comments on commit 75abe98

Please sign in to comment.