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/step.py

from typing import List

from fastapi import APIRouter, Depends, status
from fastapi.responses import JSONResponse
from pydantic import parse_obj_as

from arnold.adapter import ArnoldAdapter
from arnold.constants import QUERY_RULES
from arnold.crud import create, update
from arnold.crud.read.step import (
    aggregate_step,
    find_step_type_artifact_udfs,
    find_step_type_process_udfs,
    find_step,
    query_steps,
)
from arnold.models.database.step import Step
import logging

from arnold.models.api_models import WorkflowResponce, StepFilters, StepFiltersBase, Pagination
from arnold.settings import get_arnold_adapter
import arnold.crud.read.step

LOG = logging.getLogger(__name__)

router = APIRouter()


@router.get("/step/query_rules")
def get_query_rules():
    """Get possible filtering rules."""

    return QUERY_RULES


@router.get("/step/workflows", response_model=list[WorkflowResponce])
def get_workflows(adapter: ArnoldAdapter = Depends(get_arnold_adapter)):
    """Get available workflows and step types from the step collection"""

    pipe = [{"$group": {"_id": "$workflow", "step_types": {"$addToSet": "$step_type"}}}]
    workflows: list[dict] = aggregate_step(adapter=adapter, pipe=pipe)
    return parse_obj_as(List[WorkflowResponce], workflows)


@router.get("/step/step_type/udfs")
def get_step_type_udfs(
    step_type: str,
    workflow: str,
    adapter: ArnoldAdapter = Depends(get_arnold_adapter),
):
    """Get available artifact udfs for a step type"""

    artifact_udfs = find_step_type_artifact_udfs(
        adapter=adapter, step_type=step_type, workflow=workflow
    )
    process_udfs = find_step_type_process_udfs(
        adapter=adapter, step_type=step_type, workflow=workflow
    )
    return artifact_udfs + process_udfs


@router.get("/step/{step_id}", response_model=Step)
def get_step_by_id(
    step_id: str,
    adapter: ArnoldAdapter = Depends(get_arnold_adapter),
):
    """fetch a step by step id"""

    step: Step = find_step(step_id=step_id, adapter=adapter)
    return step


@router.post("/get_steps/", response_model=List[Step])
def get_steps(
    step_filters: StepFilters,
    adapter: ArnoldAdapter = Depends(get_arnold_adapter),
):
    """Get steps based on filters"""

    steps: List[Step] = query_steps(
        step_filters=StepFiltersBase(**step_filters.dict()),
        pagination=Pagination(**step_filters.dict()),
        udf_filters=step_filters.udf_filters,
        adapter=adapter,
    )
    return steps


@router.post("/step/")
def create_step(step: Step, adapter: ArnoldAdapter = Depends(get_arnold_adapter)) -> JSONResponse:
    if arnold.crud.read.step.find_step(step_id=step.step_id, adapter=adapter):
        return JSONResponse(
            status_code=status.HTTP_405_METHOD_NOT_ALLOWED, content="step already in database"
        )
    try:
        create.create_step(adapter=adapter, step=step)
    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"step {step.step_id} inserted to db"
    )


@router.post("/steps/")
def create_steps(
    steps: List[Step], adapter: ArnoldAdapter = Depends(get_arnold_adapter)
) -> JSONResponse:
    try:
        create.create_steps(adapter=adapter, steps=steps)
    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="steps inserted to db")


@router.put("/step/")
def update_step(step: Step, adapter: ArnoldAdapter = Depends(get_arnold_adapter)) -> JSONResponse:
    try:
        update.update_step(adapter=adapter, step=step)
    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"step {step.step_id} inserted to db"
    )


@router.put("/steps/")
def update_steps(
    steps: List[Step], adapter: ArnoldAdapter = Depends(get_arnold_adapter)
) -> JSONResponse:
    try:
        update.update_steps(adapter=adapter, steps=steps)
    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="steps inserted to db")

1	from typing import List	×
2
3	from fastapi import APIRouter, Depends, status	×
4	from fastapi.responses import JSONResponse	×
5	from pydantic import parse_obj_as	×
6
7	from arnold.adapter import ArnoldAdapter	×
8	from arnold.constants import QUERY_RULES	×
9	from arnold.crud import create, update	×
10	from arnold.crud.read.step import (	×
11	aggregate_step,
12	find_step_type_artifact_udfs,
13	find_step_type_process_udfs,
14	find_step,
15	query_steps,
16	)
17	from arnold.models.database.step import Step	×
18	import logging	×
19
20	from arnold.models.api_models import WorkflowResponce, StepFilters, StepFiltersBase, Pagination	×
21	from arnold.settings import get_arnold_adapter	×
22	import arnold.crud.read.step	×
23
24	LOG = logging.getLogger(__name__)	×
25
26	router = APIRouter()	×
27
28
29	@router.get("/step/query_rules")	×
30	def get_query_rules():	×
31	"""Get possible filtering rules."""
32
33	return QUERY_RULES	×
34
35
36	@router.get("/step/workflows", response_model=list[WorkflowResponce])	×
37	def get_workflows(adapter: ArnoldAdapter = Depends(get_arnold_adapter)):	×
38	"""Get available workflows and step types from the step collection"""
39
40	pipe = [{"$group": {"_id": "$workflow", "step_types": {"$addToSet": "$step_type"}}}]	×
41	workflows: list[dict] = aggregate_step(adapter=adapter, pipe=pipe)	×
42	return parse_obj_as(List[WorkflowResponce], workflows)	×
43
44
45	@router.get("/step/step_type/udfs")	×
46	def get_step_type_udfs(	×
47	step_type: str,
48	workflow: str,
49	adapter: ArnoldAdapter = Depends(get_arnold_adapter),
50	):
51	"""Get available artifact udfs for a step type"""
52
53	artifact_udfs = find_step_type_artifact_udfs(	×
54	adapter=adapter, step_type=step_type, workflow=workflow
55	)
56	process_udfs = find_step_type_process_udfs(	×
57	adapter=adapter, step_type=step_type, workflow=workflow
58	)
59	return artifact_udfs + process_udfs	×
60
61
62	@router.get("/step/{step_id}", response_model=Step)	×
63	def get_step_by_id(	×
64	step_id: str,
65	adapter: ArnoldAdapter = Depends(get_arnold_adapter),
66	):
67	"""fetch a step by step id"""
68
69	step: Step = find_step(step_id=step_id, adapter=adapter)	×
70	return step	×
71
72
73	@router.post("/get_steps/", response_model=List[Step])	×
74	def get_steps(	×
75	step_filters: StepFilters,
76	adapter: ArnoldAdapter = Depends(get_arnold_adapter),
77	):
78	"""Get steps based on filters"""
79
80	steps: List[Step] = query_steps(	×
81	step_filters=StepFiltersBase(**step_filters.dict()),
82	pagination=Pagination(**step_filters.dict()),
83	udf_filters=step_filters.udf_filters,
84	adapter=adapter,
85	)
86	return steps	×
87
88
89	@router.post("/step/")	×
90	def create_step(step: Step, adapter: ArnoldAdapter = Depends(get_arnold_adapter)) -> JSONResponse:	×
91	if arnold.crud.read.step.find_step(step_id=step.step_id, adapter=adapter):	×
92	return JSONResponse(	×
93	status_code=status.HTTP_405_METHOD_NOT_ALLOWED, content="step already in database"
94	)
95	try:	×
96	create.create_step(adapter=adapter, step=step)	×
97	except Exception as e:	×
98	return JSONResponse(	×
99	status_code=status.HTTP_405_METHOD_NOT_ALLOWED,
100	content=f"exception {e} ",
101	)
102
103	return JSONResponse(	×
104	status_code=status.HTTP_200_OK, content=f"step {step.step_id} inserted to db"
105	)
106
107
108	@router.post("/steps/")	×
109	def create_steps(	×
110	steps: List[Step], adapter: ArnoldAdapter = Depends(get_arnold_adapter)
111	) -> JSONResponse:
112	try:	×
113	create.create_steps(adapter=adapter, steps=steps)	×
114	except Exception as e:	×
115	return JSONResponse(	×
116	status_code=status.HTTP_405_METHOD_NOT_ALLOWED,
117	content=f"exception {e} ",
118	)
119	return JSONResponse(status_code=status.HTTP_200_OK, content="steps inserted to db")	×
120
121
122	@router.put("/step/")	×
123	def update_step(step: Step, adapter: ArnoldAdapter = Depends(get_arnold_adapter)) -> JSONResponse:	×
124	try:	×
125	update.update_step(adapter=adapter, step=step)	×
126	except Exception as e:	×
UNCOV 127	return JSONResponse(	×
128	status_code=status.HTTP_405_METHOD_NOT_ALLOWED,
129	content=f"exception {e} ",
130	)
131
UNCOV 132	return JSONResponse(	×
133	status_code=status.HTTP_200_OK, content=f"step {step.step_id} inserted to db"
134	)
135
136
137	@router.put("/steps/")	×
UNCOV 138	def update_steps(	×
139	steps: List[Step], adapter: ArnoldAdapter = Depends(get_arnold_adapter)
140	) -> JSONResponse:
141	try:	×
142	update.update_steps(adapter=adapter, steps=steps)	×
UNCOV 143	except Exception as e:	×
UNCOV 144	return JSONResponse(	×
145	status_code=status.HTTP_405_METHOD_NOT_ALLOWED,
146	content=f"exception {e} ",
147	)
UNCOV 148	return JSONResponse(status_code=status.HTTP_200_OK, content="steps 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