7102454002

Committed 05 Dec 2023 02:52PM UTC coverage: 0.0%. Remained the same

Build # 7102454002

Build Type

push

github

Committed by

web-flow

Commit Message

Update to pydantic v2 (#47) (patch)

Added
Move the models to pydantic v2

add Codeowners for automatic reviewer tagging

Run Details

0 of 89 new or added lines in 5 files covered. (0.0%)

28 existing lines in 5 files now uncovered.

0 of 699 relevant lines covered (0.0%)

0.0 hits per line

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

0.0

/arnold/api/api_v1/endpoints/sample.py

from arnold.adapter import ArnoldAdapter
from arnold.crud import create, update
from arnold.crud.read.sample import find_sample, find_all_samples
from arnold.crud.read.step import find_sample_fields
from arnold.models.database import Sample
from typing import List
from fastapi import APIRouter, Depends, status
from fastapi.responses import JSONResponse
import logging
import arnold.crud.read.sample
from arnold.settings import get_arnold_adapter

LOG = logging.getLogger(__name__)

router = APIRouter()


@router.get("/sample/fields")
def get_sample_fields(
    adapter: ArnoldAdapter = Depends(get_arnold_adapter),
):
    """Get sample fields"""
    return find_sample_fields(adapter=adapter)


@router.get("/sample/{sample_id}", response_model=Sample)
def get_sample(
    sample_id: str,
    adapter: ArnoldAdapter = Depends(get_arnold_adapter),
):
    """fetch a sample by sample id"""
    sample: Sample = find_sample(sample_id=sample_id, adapter=adapter)
    return sample


@router.get("/samples/", response_model=List[Sample])
def get_samples(
    adapter: ArnoldAdapter = Depends(get_arnold_adapter),
):
    """Get all samples"""
    samples: List[Sample] = find_all_samples(adapter=adapter)

    return samples


@router.post("/sample/")
def create_sample(
    sample: Sample, adapter: ArnoldAdapter = Depends(get_arnold_adapter)
) -> JSONResponse:
    if arnold.crud.read.sample.find_sample(sample_id=sample.sample_id, adapter=adapter):
        return JSONResponse(
            status_code=status.HTTP_405_METHOD_NOT_ALLOWED, content="Sample already in database"
        )
    try:
        create.create_sample(adapter=adapter, sample=sample)
    except Exception as e:
        return JSONResponse(
            status_code=status.HTTP_405_METHOD_NOT_ALLOWED,
            content=f"exception {e} ",
        )

    return JSONResponse(
        status_code=status.HTTP_200_OK, content=f"Sample {sample.sample_id} inserted to db"
    )


@router.post("/samples/")
def create_samples(
    samples: List[Sample], adapter: ArnoldAdapter = Depends(get_arnold_adapter)
) -> JSONResponse:
    try:
        create.create_samples(adapter=adapter, samples=samples)
    except Exception as e:
        return JSONResponse(
            status_code=status.HTTP_405_METHOD_NOT_ALLOWED,
            content=f"exception {e} ",
        )
    return JSONResponse(status_code=status.HTTP_200_OK, content="Samples inserted to db")


@router.put("/sample/")
def update_sample(
    sample: Sample, adapter: ArnoldAdapter = Depends(get_arnold_adapter)
) -> JSONResponse:
    try:
        update.update_sample(adapter=adapter, sample=sample)
    except Exception as e:
        return JSONResponse(
            status_code=status.HTTP_405_METHOD_NOT_ALLOWED,
            content=f"exception {e} ",
        )

    return JSONResponse(
        status_code=status.HTTP_200_OK, content=f"Sample {sample.sample_id} inserted to db"
    )


@router.put("/samples/")
def update_samples(
    samples: List[Sample], adapter: ArnoldAdapter = Depends(get_arnold_adapter)
) -> JSONResponse:
    try:
        update.update_samples(adapter=adapter, samples=samples)
    except Exception as e:
        return JSONResponse(
            status_code=status.HTTP_405_METHOD_NOT_ALLOWED,
            content=f"exception {e} ",
        )
    return JSONResponse(status_code=status.HTTP_200_OK, content="Samples inserted to db")

1	from arnold.adapter import ArnoldAdapter	×
2	from arnold.crud import create, update	×
3	from arnold.crud.read.sample import find_sample, find_all_samples	×
4	from arnold.crud.read.step import find_sample_fields	×
5	from arnold.models.database import Sample	×
6	from typing import List	×
7	from fastapi import APIRouter, Depends, status	×
8	from fastapi.responses import JSONResponse	×
9	import logging	×
10	import arnold.crud.read.sample	×
11	from arnold.settings import get_arnold_adapter	×
12
13	LOG = logging.getLogger(__name__)	×
14
15	router = APIRouter()	×
16
17
18	@router.get("/sample/fields")	×
19	def get_sample_fields(	×
20	adapter: ArnoldAdapter = Depends(get_arnold_adapter),
21	):
22	"""Get sample fields"""
23	return find_sample_fields(adapter=adapter)	×
24
25
26	@router.get("/sample/{sample_id}", response_model=Sample)	×
27	def get_sample(	×
28	sample_id: str,
29	adapter: ArnoldAdapter = Depends(get_arnold_adapter),
30	):
31	"""fetch a sample by sample id"""
32	sample: Sample = find_sample(sample_id=sample_id, adapter=adapter)	×
33	return sample	×
34
35
36	@router.get("/samples/", response_model=List[Sample])	×
37	def get_samples(	×
38	adapter: ArnoldAdapter = Depends(get_arnold_adapter),
39	):
40	"""Get all samples"""
41	samples: List[Sample] = find_all_samples(adapter=adapter)	×
42
43	return samples	×
44
45
46	@router.post("/sample/")	×
47	def create_sample(	×
48	sample: Sample, adapter: ArnoldAdapter = Depends(get_arnold_adapter)
49	) -> JSONResponse:
50	if arnold.crud.read.sample.find_sample(sample_id=sample.sample_id, adapter=adapter):	×
51	return JSONResponse(	×
52	status_code=status.HTTP_405_METHOD_NOT_ALLOWED, content="Sample already in database"
53	)
54	try:	×
55	create.create_sample(adapter=adapter, sample=sample)	×
56	except Exception as e:	×
57	return JSONResponse(	×
58	status_code=status.HTTP_405_METHOD_NOT_ALLOWED,
59	content=f"exception {e} ",
60	)
61
62	return JSONResponse(	×
63	status_code=status.HTTP_200_OK, content=f"Sample {sample.sample_id} inserted to db"
64	)
65
66
67	@router.post("/samples/")	×
68	def create_samples(	×
69	samples: List[Sample], adapter: ArnoldAdapter = Depends(get_arnold_adapter)
70	) -> JSONResponse:
71	try:	×
72	create.create_samples(adapter=adapter, samples=samples)	×
73	except Exception as e:	×
74	return JSONResponse(	×
75	status_code=status.HTTP_405_METHOD_NOT_ALLOWED,
76	content=f"exception {e} ",
77	)
78	return JSONResponse(status_code=status.HTTP_200_OK, content="Samples inserted to db")	×
79
80
81	@router.put("/sample/")	×
82	def update_sample(	×
83	sample: Sample, adapter: ArnoldAdapter = Depends(get_arnold_adapter)
84	) -> JSONResponse:
85	try:	×
86	update.update_sample(adapter=adapter, sample=sample)	×
87	except Exception as e:	×
UNCOV 88	return JSONResponse(	×
89	status_code=status.HTTP_405_METHOD_NOT_ALLOWED,
90	content=f"exception {e} ",
91	)
92
UNCOV 93	return JSONResponse(	×
94	status_code=status.HTTP_200_OK, content=f"Sample {sample.sample_id} inserted to db"
95	)
96
97
98	@router.put("/samples/")	×
UNCOV 99	def update_samples(	×
100	samples: List[Sample], adapter: ArnoldAdapter = Depends(get_arnold_adapter)
101	) -> JSONResponse:
102	try:	×
103	update.update_samples(adapter=adapter, samples=samples)	×
UNCOV 104	except Exception as e:	×
UNCOV 105	return JSONResponse(	×
106	status_code=status.HTTP_405_METHOD_NOT_ALLOWED,
107	content=f"exception {e} ",
108	)
UNCOV 109	return JSONResponse(status_code=status.HTTP_200_OK, content="Samples inserted to db")	×

Clinical-Genomics / arnold / 7102454002

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