app.api.result_api#

Definition of result API endpoints accessible through swagger UI.

Attributes#

Classes#

CreateDicomResult

Payload sent by the Dagster on_run_success sensor to register DICOM output.

Functions#

create_dicom_result(...)

Create a DICOM result entry after a successful Dagster reconstruction run.

create_blank_result(...)

Create a blank task result.

get_result(]) → scanhub_libraries.models.ResultOut)

Get an existing result.

get_all_task_results(...)

Get all existing results of a certain task.

delete_result(]) → None)

Delete a result.

set_result(]) → scanhub_libraries.models.ResultOut)

Update an existing result.

get_dicom(]) → starlette.responses.Response)

Serve a DICOM instance.

upload_to_xnat(]) → dict)

Upload a DICOM file to XNAT test database.

_resolve_dicom_path(result_id, filename)

Look up Result in DB and return validated path to the requested DICOM file.

_resolve_mrd_path(result_id)

Look up Result in DB and return path to the .mrd file.

get_meta(→ scanhub_libraries.models.MRDMetaResponse)

Get MRD meta info.

get_mrd_binary(protocol_id, task_id, result_id[, ids, ...])

Get MRD as binary stream.

download_mrd(protocol_id, task_id, result_id, user)

Download the full MRD file.

Module Contents#

class app.api.result_api.CreateDicomResult#

Bases: pydantic.BaseModel

Payload sent by the Dagster on_run_success sensor to register DICOM output.

directory: str#
files: list[str]#
meta: dict | None = None#
app.api.result_api.PREFIX_PATIENT_MANAGER = 'http://patient-manager:8100/api/v1/patient'#
app.api.result_api.result_router#
app.api.result_api.oauth2_scheme#
async app.api.result_api.create_dicom_result(task_id: uuid.UUID | str, payload: CreateDicomResult, user: Annotated[scanhub_libraries.models.User, Depends(get_current_user)]) scanhub_libraries.models.ResultOut#

Create a DICOM result entry after a successful Dagster reconstruction run.

async app.api.result_api.create_blank_result(task_id: str | uuid.UUID, user: Annotated[scanhub_libraries.models.User, Depends(get_current_user)]) scanhub_libraries.models.ResultOut#

Create a blank task result.

async app.api.result_api.get_result(result_id: uuid.UUID | str, user: Annotated[scanhub_libraries.models.User, Depends(get_current_user)]) scanhub_libraries.models.ResultOut#

Get an existing result.

async app.api.result_api.get_all_task_results(task_id: uuid.UUID | str, user: Annotated[scanhub_libraries.models.User, Depends(get_current_user)]) list[scanhub_libraries.models.ResultOut]#

Get all existing results of a certain task.

async app.api.result_api.delete_result(result_id: uuid.UUID | str, user: Annotated[scanhub_libraries.models.User, Depends(get_current_user)]) None#

Delete a result.

async app.api.result_api.set_result(result_id: uuid.UUID | str, payload: scanhub_libraries.models.SetResult, user: Annotated[scanhub_libraries.models.User, Depends(get_current_user)]) scanhub_libraries.models.ResultOut#

Update an existing result.

async app.api.result_api.get_dicom(protocol_id: str, task_id: str, result_id: str, filename: str, user: Annotated[scanhub_libraries.models.User, Depends(get_current_user)]) starlette.responses.Response#

Serve a DICOM instance.

async app.api.result_api.upload_to_xnat(protocol_id: str, task_id: str, result_id: str, filename: str, access_token: Annotated[str, Depends(oauth2_scheme)]) dict#

Upload a DICOM file to XNAT test database.

async app.api.result_api._resolve_dicom_path(result_id: str, filename: str)#

Look up Result in DB and return validated path to the requested DICOM file.

async app.api.result_api._resolve_mrd_path(result_id: str)#

Look up Result in DB and return path to the .mrd file.

async app.api.result_api.get_meta(protocol_id: str, task_id: str, result_id: str) scanhub_libraries.models.MRDMetaResponse#

Get MRD meta info.

async app.api.result_api.get_mrd_binary(protocol_id: str, task_id: str, result_id: str, ids: str = Query(..., description="IDs: '0,1,10-20,40-50:2'"), coil_idx: int = Query(0, ge=0, description='Coil index'), stride: int = Query(1, ge=1, description='Decimate samples by stride'))#

Get MRD as binary stream.

async app.api.result_api.download_mrd(protocol_id: str, task_id: str, result_id: str, user: Annotated[scanhub_libraries.models.User, Depends(get_current_user)])#

Download the full MRD file.