• Home
  • Features
  • Pricing
  • Docs
  • Announcements
  • Sign In

Clinical-Genomics / statina / 24180681721

09 Apr 2026 08:33AM UTC coverage: 59.483%. First build
24180681721

Pull #250

github

web-flow
New Ratio plot endpoint (#251)

Renamed endpoint

---------

Co-authored-by: islean <isak.ohlsson.angnell@gmail.com>
Pull Request #250: Integration branch for NovaSeqX changes

38 of 59 new or added lines in 9 files covered. (64.41%)

1035 of 1740 relevant lines covered (59.48%)

0.59 hits per line

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

37.21
/statina/API/v2/endpoints/sample.py
1
from datetime import datetime
1✔
2
from pathlib import Path
1✔
3
from typing import Dict, List, Optional
1✔
4

5
from fastapi import APIRouter, Depends, Form, Query, Security
1✔
6
from fastapi.encoders import jsonable_encoder
1✔
7
from fastapi.responses import FileResponse, JSONResponse
1✔
8

9
from statina.adapter import StatinaAdapter
1✔
10
from statina.API.v2.endpoints.user import get_current_active_user
1✔
11
from statina.config import get_nipt_adapter
1✔
12
from statina.constants import sample_status_options
1✔
13
from statina.crud import update
1✔
14
from statina.crud.find import samples as find_samples
1✔
15
from statina.crud.find import batches as find_batches
1✔
16
from statina.crud.find.datasets import get_dataset
1✔
17
from statina.crud.find.plots import ratio_plot_data
1✔
18
from statina.models.database import DataBaseSample, User, DatabaseBatch
1✔
19
from statina.models.database.dataset import Dataset
1✔
20
from statina.models.query_params import SamplesQuery
1✔
21
from statina.models.server.plots.ncv import Ratio131821, RatioSamples
1✔
22
from statina.models.server.sample import (
1✔
23
    SampleResponse,
24
    PaginatedSampleResponse,
25
    SampleValidator,
26
)
27

28
from statina.parse.batch import validate_file_path
1✔
29

30
router = APIRouter(prefix="/v2")
1✔
31

32

33
@router.get("/samples", response_model=PaginatedSampleResponse)
1✔
34
def samples(
1✔
35
    sample_query: SamplesQuery = Depends(SamplesQuery),
36
    current_user: User = Security(get_current_active_user, scopes=["R"]),
37
    adapter: StatinaAdapter = Depends(get_nipt_adapter),
38
):
39
    """Get samples"""
40
    database_samples: List[DataBaseSample] = find_samples.query_samples(
×
41
        **sample_query.dict(), adapter=adapter
42
    )
43
    validated_samples: List[SampleValidator] = []
×
44
    for database_sample in database_samples:
×
45
        dataset: Dataset = get_dataset(adapter=adapter, batch_id=database_sample.batch_id)
×
46
        database_sample.dataset = dataset
×
47
        validated_sample = SampleValidator(
×
48
            **database_sample.dict(),
49
        )
50
        validated_samples.append(validated_sample)
×
51
    samples: List[SampleResponse] = [
×
52
        SampleResponse(**sample.dict()) for sample in validated_samples
53
    ]
54

55
    document_count: int = find_samples.count_query_samples(
×
56
        adapter=adapter, batch_id=sample_query.batch_id, query_string=sample_query.query_string
57
    )
58
    return JSONResponse(
×
59
        content=jsonable_encoder(
60
            PaginatedSampleResponse(document_count=document_count, documents=samples), by_alias=True
61
        )
62
    )
63

64

65
@router.get("/sample/{sample_id}", response_model=SampleResponse)
1✔
66
def sample(
1✔
67
    sample_id: str,
68
    current_user: User = Security(get_current_active_user, scopes=["R"]),
69
    adapter: StatinaAdapter = Depends(get_nipt_adapter),
70
):
71
    """Get sample with id"""
72

73
    database_sample: DataBaseSample = find_samples.sample(sample_id=sample_id, adapter=adapter)
×
74
    batch: DatabaseBatch = find_batches.batch(batch_id=database_sample.batch_id, adapter=adapter)
×
75

76
    if not database_sample:
×
77
        return JSONResponse("Not found", status_code=404)
×
78

79
    database_sample.dataset = get_dataset(adapter=adapter, batch_id=database_sample.batch_id)
×
80
    validated_sample = SampleValidator(**database_sample.dict())
×
81
    sample_view_data = SampleResponse(
×
82
        sequencing_date=batch.SequencingDate,
83
        **validated_sample.dict(exclude_none=True),
84
    )
85

86
    return JSONResponse(
×
87
        content=jsonable_encoder(
88
            sample_view_data,
89
            by_alias=True,
90
        )
91
    )
92

93

94
@router.get("/sample/{sample_id}/tris")
1✔
95
def sample_tris(
1✔
96
    sample_id: str,
97
    current_user: User = Security(get_current_active_user, scopes=["R"]),
98
    adapter: StatinaAdapter = Depends(get_nipt_adapter),
99
):
100
    """Sample view with trisomi plot."""
101

102
    database_sample: DataBaseSample = find_samples.sample(sample_id=sample_id, adapter=adapter)
×
NEW
103
    abnormal_data: Dict[str, RatioSamples] = ratio_plot_data.get_abn_for_samp_tris_plot(
×
104
        adapter=adapter
105
    )
NEW
106
    normal_data: Ratio131821 = ratio_plot_data.get_normal_for_samp_tris_plot(adapter=adapter)
×
NEW
107
    sample_data: RatioSamples = ratio_plot_data.get_sample_for_samp_tris_plot(database_sample)
×
108

109
    return JSONResponse(
×
110
        content=jsonable_encoder(
111
            dict(
112
                normal_data=normal_data.dict(exclude_none=True, by_alias=True),
113
                abnormal_data=abnormal_data,
114
                sample_data=sample_data,
115
            ),
116
            by_alias=False,
117
        ),
118
    )
119

120

121
@router.put("/sample/{sample_id}/status/13")
1✔
122
async def set_sample_status_13(
1✔
123
    sample_id: str,
124
    status: sample_status_options = Query(...),
125
    adapter: StatinaAdapter = Depends(get_nipt_adapter),
126
    current_user: User = Security(get_current_active_user, scopes=["RW"]),
127
):
128
    """Update the manualy interpreted chromosome abnormality status for a sample."""
129

130
    sample: DataBaseSample = find_samples.sample(sample_id=sample_id, adapter=adapter)
×
131
    sample.status_13 = status
×
132
    time_stamp: str = datetime.now().strftime("%Y/%m/%d %H:%M:%S")
×
133
    sample.status_change_13 = f"{current_user.username} {time_stamp}"
×
134
    update.sample(adapter=adapter, sample=sample)
×
135
    return JSONResponse(content="Sample field updated", status_code=200)
×
136

137

138
@router.put("/sample/{sample_id}/status/18")
1✔
139
async def set_sample_status_18(
1✔
140
    sample_id: str,
141
    status: sample_status_options = Query(...),
142
    adapter: StatinaAdapter = Depends(get_nipt_adapter),
143
    current_user: User = Security(get_current_active_user, scopes=["RW"]),
144
):
145
    """Update the manualy interpreted chromosome abnormality status for a sample."""
146

147
    sample: DataBaseSample = find_samples.sample(sample_id=sample_id, adapter=adapter)
×
148
    sample.status_18 = status
×
149
    time_stamp: str = datetime.now().strftime("%Y/%m/%d %H:%M:%S")
×
150
    sample.status_change_18 = f"{current_user.username} {time_stamp}"
×
151
    update.sample(adapter=adapter, sample=sample)
×
152
    return JSONResponse(content="Sample field updated", status_code=200)
×
153

154

155
@router.put("/sample/{sample_id}/status/21")
1✔
156
async def set_sample_status_21(
1✔
157
    sample_id: str,
158
    status: sample_status_options = Query(...),
159
    adapter: StatinaAdapter = Depends(get_nipt_adapter),
160
    current_user: User = Security(get_current_active_user, scopes=["RW"]),
161
):
162
    """Update the manualy interpreted chromosome abnormality status for a sample."""
163

164
    sample: DataBaseSample = find_samples.sample(sample_id=sample_id, adapter=adapter)
×
165
    sample.status_21 = status
×
166
    time_stamp: str = datetime.now().strftime("%Y/%m/%d %H:%M:%S")
×
167
    sample.status_change_21 = f"{current_user.username} {time_stamp}"
×
168
    update.sample(adapter=adapter, sample=sample)
×
169
    return JSONResponse(content="Sample field updated", status_code=200)
×
170

171

172
@router.put("/sample/{sample_id}/status/x0")
1✔
173
async def set_sample_status_x0(
1✔
174
    sample_id: str,
175
    status: sample_status_options = Query(...),
176
    adapter: StatinaAdapter = Depends(get_nipt_adapter),
177
    current_user: User = Security(get_current_active_user, scopes=["RW"]),
178
):
179
    """Update the manualy interpreted chromosome abnormality status for a sample."""
180

181
    sample: DataBaseSample = find_samples.sample(sample_id=sample_id, adapter=adapter)
×
182
    sample.status_X0 = status
×
183
    time_stamp: str = datetime.now().strftime("%Y/%m/%d %H:%M:%S")
×
184
    sample.status_change_X0 = f"{current_user.username} {time_stamp}"
×
185
    update.sample(adapter=adapter, sample=sample)
×
186
    return JSONResponse(content="Sample field updated", status_code=200)
×
187

188

189
@router.put("/sample/{sample_id}/status/xxy")
1✔
190
async def set_sample_status_xxy(
1✔
191
    sample_id: str,
192
    status: sample_status_options = Query(...),
193
    adapter: StatinaAdapter = Depends(get_nipt_adapter),
194
    current_user: User = Security(get_current_active_user, scopes=["RW"]),
195
):
196
    """Update the manualy interpreted chromosome abnormality status for a sample."""
197

198
    sample: DataBaseSample = find_samples.sample(sample_id=sample_id, adapter=adapter)
×
199
    sample.status_XXY = status
×
200
    time_stamp: str = datetime.now().strftime("%Y/%m/%d %H:%M:%S")
×
201
    sample.status_change_XXY = f"{current_user.username} {time_stamp}"
×
202
    update.sample(adapter=adapter, sample=sample)
×
203
    return JSONResponse(content="Sample field updated", status_code=200)
×
204

205

206
@router.put("/sample/{sample_id}/status/xxx")
1✔
207
async def set_sample_status_xxx(
1✔
208
    sample_id: str,
209
    status: sample_status_options = Query(...),
210
    adapter: StatinaAdapter = Depends(get_nipt_adapter),
211
    current_user: User = Security(get_current_active_user, scopes=["RW"]),
212
):
213
    """Update the manualy interpreted chromosome abnormality status for a sample."""
214

215
    sample: DataBaseSample = find_samples.sample(sample_id=sample_id, adapter=adapter)
×
216
    sample.status_XXX = status
×
217
    time_stamp: str = datetime.now().strftime("%Y/%m/%d %H:%M:%S")
×
218
    sample.status_change_XXX = f"{current_user.username} {time_stamp}"
×
219
    update.sample(adapter=adapter, sample=sample)
×
220
    return JSONResponse(content="Sample field updated", status_code=200)
×
221

222

223
@router.put("/sample/{sample_id}/status/xyy")
1✔
224
async def set_sample_status_xyy(
1✔
225
    sample_id: str,
226
    status: sample_status_options = Query(...),
227
    adapter: StatinaAdapter = Depends(get_nipt_adapter),
228
    current_user: User = Security(get_current_active_user, scopes=["RW"]),
229
):
230
    """Update the manualy interpreted chromosome abnormality status for a sample."""
231

232
    sample: DataBaseSample = find_samples.sample(sample_id=sample_id, adapter=adapter)
×
233
    sample.status_XYY = status
×
234
    time_stamp: str = datetime.now().strftime("%Y/%m/%d %H:%M:%S")
×
235
    sample.status_change_XYY = f"{current_user.username} {time_stamp}"
×
236
    update.sample(adapter=adapter, sample=sample)
×
237
    return JSONResponse(content="Sample field updated", status_code=200)
×
238

239

240
@router.put("/sample/{sample_id}/comment")
1✔
241
async def sample_comment(
1✔
242
    sample_id: str,
243
    comment: Optional[str] = Form(...),
244
    adapter: StatinaAdapter = Depends(get_nipt_adapter),
245
    current_user: User = Security(get_current_active_user, scopes=["RW"]),
246
):
247
    """Update sample comment"""
248

249
    sample: DataBaseSample = find_samples.sample(sample_id=sample_id, adapter=adapter)
×
250
    sample.comment = comment
×
251
    update.sample(adapter=adapter, sample=sample)
×
252

253
    return JSONResponse(content="Sample comment updated", status_code=200)
×
254

255

256
@router.put("/sample/{sample_id}/include")
1✔
257
async def sample_include(
1✔
258
    sample_id: str,
259
    include: bool = Query(...),
260
    adapter: StatinaAdapter = Depends(get_nipt_adapter),
261
    current_user: User = Security(get_current_active_user, scopes=["RW"]),
262
):
263
    """Include sample in plots"""
264

265
    sample: DataBaseSample = find_samples.sample(sample_id=sample_id, adapter=adapter)
×
266
    sample.include = include
×
267
    update.sample(adapter=adapter, sample=sample)
×
268

269
    return JSONResponse(content="Sample inclusion status updated", status_code=200)
×
270

271

272
@router.get("/sample/{sample_id}/download/segmental_calls")
1✔
273
def sample_segmental_calls_download(
1✔
274
    sample_id: str,
275
    current_user: User = Security(get_current_active_user, scopes=["R"]),
276
    adapter: StatinaAdapter = Depends(get_nipt_adapter),
277
):
278
    """View for sample downloads"""
279

280
    sample: DataBaseSample = find_samples.sample(adapter=adapter, sample_id=sample_id)
×
281
    file_path = sample.dict().get("segmental_calls")
×
282
    if not validate_file_path(file_path):
×
283
        # warn file missing!
284
        JSONResponse(content="File missing on disk", status_code=404)
×
285

286
    file = Path(file_path)
×
287
    response = FileResponse(
×
288
        str(file.absolute()), media_type="application/octet-stream", filename=file.name
289
    )
290
    response.headers["Access-Control-Expose-Headers"] = "Content-Disposition"
×
291
    return response
×
STATUS · Troubleshooting · Open an Issue · Sales · Support · CAREERS · ENTERPRISE · START FREE · SCHEDULE DEMO
ANNOUNCEMENTS · TWITTER · TOS & SLA · Supported CI Services · What's a CI service? · Automated Testing

© 2026 Coveralls, Inc