8169178037

Committed 06 Mar 2024 08:39AM UTC 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/analyses.py

"""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 import get_session
from genotype_api.file_parsing.files import check_file
from genotype_api.models import Analysis, AnalysisRead, AnalysisReadWithGenotype, User
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

SelectOfScalar.inherit_cache = True
Select.inherit_cache = True

router = APIRouter()


@router.get("/{analysis_id}", response_model=AnalysisReadWithGenotype)
def read_analysis(
    analysis_id: int,
    session: Session = Depends(get_session),
    current_user: User = Depends(get_active_user),
):
    """Return analysis."""

    return get_analysis(session=session, analysis_id=analysis_id)


@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]:
    """Return all analyses."""
    analyses: List[Analysis] = session.exec(select(Analysis).offset(skip).limit(limit)).all()

    return analyses


@router.delete("/{analysis_id}")
def delete_analysis(
    analysis_id: int,
    session: Session = Depends(get_session),
    current_user: User = Depends(get_active_user),
):
    """Delete analysis based on analysis_id"""
    analysis = get_analysis(session=session, analysis_id=analysis_id)
    session.delete(analysis)
    session.commit()

    return JSONResponse(f"Deleted analysis: {analysis_id}", status_code=status.HTTP_200_OK)


@router.post("/sequence", response_model=List[Analysis])
def upload_sequence_analysis(
    file: UploadFile = File(...),
    session: Session = Depends(get_session),
    current_user: User = Depends(get_active_user),
):
    """Reading vcf file, creating and uploading sequence analyses and sample objects to db"""

    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())
    check_analyses_objects(session=session, analyses=analyses, analysis_type="sequence")
    create_analyses_sample_objects(session=session, analyses=analyses)
    for analysis in analyses:
        analysis: Analysis = create_analysis(session=session, analysis=analysis)
        refresh_sample_status(session=session, sample=analysis.sample)
    return analyses

1	"""Routes for analysis"""
2
3	from pathlib import Path	×
4	from typing import List	×
5
6	from fastapi import APIRouter, Depends, status, Query, UploadFile, File	×
7	from fastapi.responses import JSONResponse	×
8
9	from genotype_api.crud.analyses import (	×
10	get_analysis,
11	check_analyses_objects,
12	create_analysis,
13	)
14	from genotype_api.crud.samples import (	×
15	create_analyses_sample_objects,
16	refresh_sample_status,
17	)
18	from genotype_api.database import get_session	×
19	from genotype_api.file_parsing.files import check_file	×
20	from genotype_api.models import Analysis, AnalysisRead, AnalysisReadWithGenotype, User	×
21	from sqlmodel import Session, select	×
22
23	from genotype_api.security import get_active_user	×
24	from genotype_api.file_parsing.vcf import SequenceAnalysis	×
25	from sqlmodel.sql.expression import Select, SelectOfScalar	×
26
27	SelectOfScalar.inherit_cache = True	×
28	Select.inherit_cache = True	×
29
30	router = APIRouter()	×
31
32
33	@router.get("/{analysis_id}", response_model=AnalysisReadWithGenotype)	×
34	def read_analysis(	×
35	analysis_id: int,
36	session: Session = Depends(get_session),
37	current_user: User = Depends(get_active_user),
38	):
39	"""Return analysis."""
40
41	return get_analysis(session=session, analysis_id=analysis_id)	×
42
43
44	@router.get("/", response_model=List[AnalysisRead])	×
45	def read_analyses(	×
46	skip: int = 0,
47	limit: int = Query(default=100, lte=100),
48	session: Session = Depends(get_session),
49	current_user: User = Depends(get_active_user),
50	) -> List[Analysis]:
51	"""Return all analyses."""
52	analyses: List[Analysis] = session.exec(select(Analysis).offset(skip).limit(limit)).all()	×
53
54	return analyses	×
55
56
57	@router.delete("/{analysis_id}")	×
58	def delete_analysis(	×
59	analysis_id: int,
60	session: Session = Depends(get_session),
61	current_user: User = Depends(get_active_user),
62	):
63	"""Delete analysis based on analysis_id"""
64	analysis = get_analysis(session=session, analysis_id=analysis_id)	×
65	session.delete(analysis)	×
66	session.commit()	×
67
68	return JSONResponse(f"Deleted analysis: {analysis_id}", status_code=status.HTTP_200_OK)	×
69
70
71	@router.post("/sequence", response_model=List[Analysis])	×
72	def upload_sequence_analysis(	×
73	file: UploadFile = File(...),
74	session: Session = Depends(get_session),
75	current_user: User = Depends(get_active_user),
76	):
77	"""Reading vcf file, creating and uploading sequence analyses and sample objects to db"""
78
79	file_name: Path = check_file(file_path=file.filename, extension=".vcf")	×
80	content = file.file.read().decode("utf-8")	×
81	sequence_analysis = SequenceAnalysis(vcf_file=content, source=str(file_name))	×
82	analyses: List[Analysis] = list(sequence_analysis.generate_analyses())	×
83	check_analyses_objects(session=session, analyses=analyses, analysis_type="sequence")	×
84	create_analyses_sample_objects(session=session, analyses=analyses)	×
85	for analysis in analyses:	×
86	analysis: Analysis = create_analysis(session=session, analysis=analysis)	×
87	refresh_sample_status(session=session, sample=analysis.sample)	×
88	return analyses	×

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