8169178037

Committed 06 Mar 2024 08:39AM CUT coverage: 0.106%. First build

Build # 8169178037

Build Type

Pull #79

github

Committed by

henrikstranneheim

Commit Message

Merge branch 'main' into add-black

# Conflicts:
#	requirements.txt

Pull Request Pull Request #79: feat(black)

Run Details

1 of 940 relevant lines covered (0.11%)

0.0 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

0.0

/genotype_api/api/endpoints/samples.py

from typing import List, Optional, Literal
from fastapi import APIRouter, Depends, Query
from fastapi.responses import JSONResponse
from datetime import datetime, timedelta, date
from starlette import status

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,
    SampleReadWithAnalysisDeep,
    compare_genotypes,
)
from collections import Counter
from genotype_api import crud
from genotype_api.crud.samples import (
    get_incomplete_samples,
    get_plate_samples,
    get_commented_samples,
    get_sample,
    get_status_missing_samples,
    refresh_sample_status,
    get_samples,
)
from sqlmodel import Session, select
from sqlmodel.sql.expression import Select, SelectOfScalar
from genotype_api.security import get_active_user

SelectOfScalar.inherit_cache = True
Select.inherit_cache = True


router = APIRouter()


@router.get(
    "/{sample_id}",
    response_model=SampleReadWithAnalysisDeep,
    response_model_by_alias=False,
    response_model_exclude={
        "analyses": {"__all__": {"genotypes": True, "source": True, "created_at": True}},
        "detail": {
            "sex": True,
            "nocalls": True,
            "snps": True,
            "matches": True,
            "mismatches": True,
            "unknown": True,
        },
    },
)
def read_sample(
    sample_id: str,
    session: Session = Depends(get_session),
    current_user: User = Depends(get_active_user),
):
    sample: Sample = get_sample(session=session, sample_id=sample_id)
    if len(sample.analyses) == 2 and not sample.status:
        sample: Sample = refresh_sample_status(session=session, sample=sample)
    return sample


@router.get(
    "/",
    response_model=List[SampleReadWithAnalysisDeep],
    response_model_by_alias=False,
    response_model_exclude={
        "analyses": {"__all__": {"genotypes": True, "source": True, "created_at": True}},
        "detail": {
            "sex": True,
            "nocalls": True,
            "snps": True,
            "matches": True,
            "mismatches": True,
            "unknown": True,
        },
    },
)
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,
    session: Session = Depends(get_session),
    current_user: User = Depends(get_active_user),
) -> List[Sample]:
    """Returns a list of samples matching the provided filters."""
    statement: SelectOfScalar = select(Sample).distinct().join(Analysis)
    if sample_id:
        statement: SelectOfScalar = get_samples(statement=statement, sample_id=sample_id)
    if plate_id:
        statement: SelectOfScalar = get_plate_samples(statement=statement, plate_id=plate_id)
    if incomplete:
        statement: SelectOfScalar = get_incomplete_samples(statement=statement)
    if commented:
        statement: SelectOfScalar = get_commented_samples(statement=statement)
    if status_missing:
        statement: SelectOfScalar = get_status_missing_samples(statement=statement)
    return session.exec(
        statement.order_by(Sample.created_at.desc()).offset(skip).limit(limit)
    ).all()


@router.post("/", response_model=SampleRead)
def create_sample(
    sample: Sample,
    session: Session = Depends(get_session),
    current_user: User = Depends(get_active_user),
):
    return crud.samples.create_sample(session=session, sample=sample)


@router.put("/{sample_id}/sex", response_model=SampleRead)
def update_sex(
    sample_id: str,
    sex: SEXES = Query(...),
    genotype_sex: Optional[SEXES] = None,
    sequence_sex: Optional[SEXES] = None,
    session: Session = Depends(get_session),
    current_user: User = Depends(get_active_user),
):
    """Updating sex field on sample and sample analyses."""

    sample_in_db: Sample = get_sample(session=session, sample_id=sample_id)
    sample_in_db.sex = sex
    for analysis in sample_in_db.analyses:
        if genotype_sex and analysis.type == "genotype":
            analysis.sex = genotype_sex
        elif sequence_sex and analysis.type == "sequence":
            analysis.sex = sequence_sex
        session.add(analysis)
    session.add(sample_in_db)
    session.commit()
    session.refresh(sample_in_db)
    sample_in_db: Sample = refresh_sample_status(session=session, sample=sample_in_db)
    return sample_in_db


@router.put("/{sample_id}/comment", response_model=SampleRead)
def update_comment(
    sample_id: str,
    comment: str = Query(...),
    session: Session = Depends(get_session),
    current_user: User = Depends(get_active_user),
):
    """Updating comment field on sample."""

    sample_in_db: Sample = get_sample(session=session, sample_id=sample_id)
    sample_in_db.comment = comment
    session.add(sample_in_db)
    session.commit()
    session.refresh(sample_in_db)
    return sample_in_db


@router.put("/{sample_id}/status", response_model=SampleRead)
def set_sample_status(
    sample_id: str,
    session: Session = Depends(get_session),
    status: Optional[Literal["pass", "fail", "cancel"]] = None,
    current_user: User = Depends(get_active_user),
):
    """Check sample analyses and update sample status accordingly."""

    sample: Sample = get_sample(session=session, sample_id=sample_id)
    sample.status = status
    session.add(sample)
    session.commit()
    session.refresh(sample)
    return sample


@router.get("/{sample_id}/match", response_model=List[MatchResult])
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,
    session: Session = Depends(get_session),
    current_user: User = Depends(get_active_user),
) -> List[MatchResult]:
    """Match sample genotype against all other genotypes."""

    all_genotypes: Analysis = session.query(Analysis).filter(
        Analysis.type == comparison_set,
        Analysis.created_at > date_min - timedelta(days=1),
        Analysis.created_at < date_max + timedelta(days=1),
    )
    genotype_checked = (
        session.query(Analysis).filter(
            Analysis.sample_id == sample_id, Analysis.type == analysis_type
        )
    ).one()

    match_results = []
    for genotype in all_genotypes:
        genotype_pairs = zip(genotype.genotypes, genotype_checked.genotypes)
        results = dict(
            compare_genotypes(genotype_1, genotype_2) for genotype_1, genotype_2 in genotype_pairs
        )
        count = Counter([val for key, val in results.items()])
        if count.get("match", 0) + count.get("unknown", 0) > 40:
            match_results.append(
                MatchResult(
                    sample_id=genotype.sample_id,
                    match_results=MatchCounts.parse_obj(count),
                ),
            )
    return match_results


@router.get(
    "/{sample_id}/status_detail",
    response_model=SampleDetail,
    deprecated=True,
    response_model_include={"sex": True, "nocalls": True, "snps": True},
)
def get_status_detail(
    sample_id: str,
    session: Session = Depends(get_session),
    current_user: User = Depends(get_active_user),
):
    sample: Sample = get_sample(session=session, sample_id=sample_id)
    if len(sample.analyses) != 2:
        return SampleDetail()
    return check_sample(sample=sample)


@router.delete("/{sample_id}", response_model=Sample)
def delete_sample(
    sample_id: str,
    session: Session = Depends(get_session),
    current_user: User = Depends(get_active_user),
):
    """Delete sample and its Analyses."""

    sample: Sample = get_sample(session=session, sample_id=sample_id)
    for analysis in sample.analyses:
        session.delete(analysis)
    session.delete(sample)
    session.commit()
    return JSONResponse("Deleted", status_code=status.HTTP_200_OK)

1	from typing import List, Optional, Literal	×
2	from fastapi import APIRouter, Depends, Query	×
3	from fastapi.responses import JSONResponse	×
4	from datetime import datetime, timedelta, date	×
5	from starlette import status	×
6
7	from genotype_api.constants import SEXES	×
8	from genotype_api.database import get_session	×
9	from genotype_api.match import check_sample	×
10	from genotype_api.models import (	×
11	Sample,
12	SampleReadWithAnalysis,
13	SampleRead,
14	User,
15	SampleDetail,
16	Analysis,
17	MatchResult,
18	MatchCounts,
19	SampleReadWithAnalysisDeep,
20	compare_genotypes,
21	)
22	from collections import Counter	×
23	from genotype_api import crud	×
24	from genotype_api.crud.samples import (	×
25	get_incomplete_samples,
26	get_plate_samples,
27	get_commented_samples,
28	get_sample,
29	get_status_missing_samples,
30	refresh_sample_status,
31	get_samples,
32	)
33	from sqlmodel import Session, select	×
34	from sqlmodel.sql.expression import Select, SelectOfScalar	×
35	from genotype_api.security import get_active_user	×
36
37	SelectOfScalar.inherit_cache = True	×
38	Select.inherit_cache = True	×
39
40
41	router = APIRouter()	×
42
43
44	@router.get(	×
45	"/{sample_id}",
46	response_model=SampleReadWithAnalysisDeep,
47	response_model_by_alias=False,
48	response_model_exclude={
49	"analyses": {"__all__": {"genotypes": True, "source": True, "created_at": True}},
50	"detail": {
51	"sex": True,
52	"nocalls": True,
53	"snps": True,
54	"matches": True,
55	"mismatches": True,
56	"unknown": True,
57	},
58	},
59	)
60	def read_sample(	×
61	sample_id: str,
62	session: Session = Depends(get_session),
63	current_user: User = Depends(get_active_user),
64	):
65	sample: Sample = get_sample(session=session, sample_id=sample_id)	×
66	if len(sample.analyses) == 2 and not sample.status:	×
67	sample: Sample = refresh_sample_status(session=session, sample=sample)	×
68	return sample	×
69
70
71	@router.get(	×
72	"/",
73	response_model=List[SampleReadWithAnalysisDeep],
74	response_model_by_alias=False,
75	response_model_exclude={
76	"analyses": {"__all__": {"genotypes": True, "source": True, "created_at": True}},
77	"detail": {
78	"sex": True,
79	"nocalls": True,
80	"snps": True,
81	"matches": True,
82	"mismatches": True,
83	"unknown": True,
84	},
85	},
86	)
87	def read_samples(	×
88	skip: int = 0,
89	limit: int = Query(default=10, lte=10),
90	sample_id: Optional[str] = None,
91	plate_id: Optional[str] = None,
92	incomplete: Optional[bool] = False,
93	commented: Optional[bool] = False,
94	status_missing: Optional[bool] = False,
95	session: Session = Depends(get_session),
96	current_user: User = Depends(get_active_user),
97	) -> List[Sample]:
98	"""Returns a list of samples matching the provided filters."""
99	statement: SelectOfScalar = select(Sample).distinct().join(Analysis)	×
100	if sample_id:	×
101	statement: SelectOfScalar = get_samples(statement=statement, sample_id=sample_id)	×
102	if plate_id:	×
103	statement: SelectOfScalar = get_plate_samples(statement=statement, plate_id=plate_id)	×
104	if incomplete:	×
105	statement: SelectOfScalar = get_incomplete_samples(statement=statement)	×
106	if commented:	×
107	statement: SelectOfScalar = get_commented_samples(statement=statement)	×
108	if status_missing:	×
109	statement: SelectOfScalar = get_status_missing_samples(statement=statement)	×
110	return session.exec(	×
111	statement.order_by(Sample.created_at.desc()).offset(skip).limit(limit)
112	).all()
113
114
115	@router.post("/", response_model=SampleRead)	×
116	def create_sample(	×
117	sample: Sample,
118	session: Session = Depends(get_session),
119	current_user: User = Depends(get_active_user),
120	):
121	return crud.samples.create_sample(session=session, sample=sample)	×
122
123
124	@router.put("/{sample_id}/sex", response_model=SampleRead)	×
125	def update_sex(	×
126	sample_id: str,
127	sex: SEXES = Query(...),
128	genotype_sex: Optional[SEXES] = None,
129	sequence_sex: Optional[SEXES] = None,
130	session: Session = Depends(get_session),
131	current_user: User = Depends(get_active_user),
132	):
133	"""Updating sex field on sample and sample analyses."""
134
135	sample_in_db: Sample = get_sample(session=session, sample_id=sample_id)	×
136	sample_in_db.sex = sex	×
137	for analysis in sample_in_db.analyses:	×
138	if genotype_sex and analysis.type == "genotype":	×
139	analysis.sex = genotype_sex	×
140	elif sequence_sex and analysis.type == "sequence":	×
141	analysis.sex = sequence_sex	×
142	session.add(analysis)	×
143	session.add(sample_in_db)	×
144	session.commit()	×
145	session.refresh(sample_in_db)	×
146	sample_in_db: Sample = refresh_sample_status(session=session, sample=sample_in_db)	×
147	return sample_in_db	×
148
149
150	@router.put("/{sample_id}/comment", response_model=SampleRead)	×
151	def update_comment(	×
152	sample_id: str,
153	comment: str = Query(...),
154	session: Session = Depends(get_session),
155	current_user: User = Depends(get_active_user),
156	):
157	"""Updating comment field on sample."""
158
159	sample_in_db: Sample = get_sample(session=session, sample_id=sample_id)	×
160	sample_in_db.comment = comment	×
161	session.add(sample_in_db)	×
162	session.commit()	×
163	session.refresh(sample_in_db)	×
164	return sample_in_db	×
165
166
167	@router.put("/{sample_id}/status", response_model=SampleRead)	×
168	def set_sample_status(	×
169	sample_id: str,
170	session: Session = Depends(get_session),
171	status: Optional[Literal["pass", "fail", "cancel"]] = None,
172	current_user: User = Depends(get_active_user),
173	):
174	"""Check sample analyses and update sample status accordingly."""
175
176	sample: Sample = get_sample(session=session, sample_id=sample_id)	×
177	sample.status = status	×
178	session.add(sample)	×
179	session.commit()	×
180	session.refresh(sample)	×
181	return sample	×
182
183
184	@router.get("/{sample_id}/match", response_model=List[MatchResult])	×
185	def match(	×
186	sample_id: str,
187	analysis_type: Literal["genotype", "sequence"],
188	comparison_set: Literal["genotype", "sequence"],
189	date_min: Optional[date] = date.min,
190	date_max: Optional[date] = date.max,
191	session: Session = Depends(get_session),
192	current_user: User = Depends(get_active_user),
193	) -> List[MatchResult]:
194	"""Match sample genotype against all other genotypes."""
195
196	all_genotypes: Analysis = session.query(Analysis).filter(	×
197	Analysis.type == comparison_set,
198	Analysis.created_at > date_min - timedelta(days=1),
199	Analysis.created_at < date_max + timedelta(days=1),
200	)
201	genotype_checked = (	×
202	session.query(Analysis).filter(
203	Analysis.sample_id == sample_id, Analysis.type == analysis_type
204	)
205	).one()
206
207	match_results = []	×
208	for genotype in all_genotypes:	×
209	genotype_pairs = zip(genotype.genotypes, genotype_checked.genotypes)	×
210	results = dict(	×
211	compare_genotypes(genotype_1, genotype_2) for genotype_1, genotype_2 in genotype_pairs
212	)
213	count = Counter([val for key, val in results.items()])	×
214	if count.get("match", 0) + count.get("unknown", 0) > 40:	×
215	match_results.append(	×
216	MatchResult(
217	sample_id=genotype.sample_id,
218	match_results=MatchCounts.parse_obj(count),
219	),
220	)
221	return match_results	×
222
223
224	@router.get(	×
225	"/{sample_id}/status_detail",
226	response_model=SampleDetail,
227	deprecated=True,
228	response_model_include={"sex": True, "nocalls": True, "snps": True},
229	)
230	def get_status_detail(	×
231	sample_id: str,
232	session: Session = Depends(get_session),
233	current_user: User = Depends(get_active_user),
234	):
235	sample: Sample = get_sample(session=session, sample_id=sample_id)	×
236	if len(sample.analyses) != 2:	×
237	return SampleDetail()	×
238	return check_sample(sample=sample)	×
239
240
241	@router.delete("/{sample_id}", response_model=Sample)	×
242	def delete_sample(	×
243	sample_id: str,
244	session: Session = Depends(get_session),
245	current_user: User = Depends(get_active_user),
246	):
247	"""Delete sample and its Analyses."""
248
249	sample: Sample = get_sample(session=session, sample_id=sample_id)	×
250	for analysis in sample.analyses:	×
251	session.delete(analysis)	×
252	session.delete(sample)	×
253	session.commit()	×
254	return JSONResponse("Deleted", status_code=status.HTTP_200_OK)	×

Clinical-Genomics / genotype-api / 8169178037

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous