The return of the Python LORIS API client

pyproject.toml

@@ -6,7 +6,7 @@ line-length = 120
 preview = true
 
 [tool.ruff.lint]
-ignore = ["E202", "E203", "E221", "E241", "E251", "E272"]
+ignore = ["E202", "E203", "E221", "E241", "E251", "E271","E272"]
 select = ["E", "F", "I", "N", "UP", "W"]
 
 # The strict type checking configuration is used to type check only the modern (typed) modules. An
@@ -15,6 +15,7 @@ select = ["E", "F", "I", "N", "UP", "W"]
 [tool.pyright]
 include = [
     "python/tests",
+    "python/lib/api",
     "python/lib/db",
     "python/lib/exception",
     "python/lib/config_file.py",

python/lib/api/client.py

@@ -0,0 +1,100 @@
+from dataclasses import dataclass
+from typing import Any, Literal
+
+import requests
+from requests import HTTPError
+
+# TODO: Turn into a type declaration with Python 3.12
+ApiVersion = Literal['v0.0.3', 'v0.0.4-dev']
+
+
+@dataclass
+class ApiClient:
+    loris_url: str
+    api_token: str
+
+    def get(
+        self,
+        version: ApiVersion,
+        route: str,
+        json: dict[str, Any] | None = None,
+    ):
+        headers = {
+            'Authorization': f'Bearer {self.api_token}',
+        }
+
+        try:
+            response = requests.get(
+                f'https://{self.loris_url}/api/{version}/{route}',
+                headers=headers,
+                json=json,
+                allow_redirects=False,
+            )
+
+            response.raise_for_status()
+            return response
+        except HTTPError as error:
+            # TODO: Better error handling
+            print(error.response.status_code)
+            print(error.response.text)
+            exit(0)
+
+    def post(
+        self,
+        version: ApiVersion,
+        route: str,
+        data: dict[str, str] = {},
+        json: dict[str, Any] | None = None,
+        files: dict[str, Any] | None = None,
+    ):
+        headers = {
+            'Authorization': f'Bearer {self.api_token}',
+        }
+
+        try:
+            response = requests.post(
+                f'https://{self.loris_url}/api/{version}/{route}',
+                headers=headers,
+                data=data,
+                json=json,
+                files=files,
+                allow_redirects=False,
+            )
+
+            response.raise_for_status()
+            return response
+        except HTTPError as error:
+            # TODO: Better error handling
+            print(error.response.status_code)
+            print(error.response.text)
+            exit(0)
+
+
+def get_api_token(loris_url: str, username: str, password: str) -> str:
+    """
+    Call the LORIS API to get an API token for a given LORIS user using this user's credentials.
+    """
+
+    credentials = {
+        'username': username,
+        'password': password,
+    }
+
+    try:
+        response = requests.post(f'https://{loris_url}/api/v0.0.4-dev/login', json=credentials)
+        response.raise_for_status()
+        return response.json()['token']
+    except HTTPError as error:
+        error_description = error.response.json()['error']
+        if error_description == 'Unacceptable JWT key':
+            raise Exception(
+                'Unacceptable LORIS JWT key.\n'
+                'To use the API, please enter a sufficiently complex JWT key in the LORIS configuration module.'
+            )
+
+        exit(0)
+
+
+def get_api_client(loris_url: str, username: str, password: str):
+    api_token = get_api_token(loris_url, username, password)
+    return ApiClient(loris_url, api_token)

python/lib/api/endpoints/dicom.py

@@ -0,0 +1,67 @@
+import json
+import os
+
+from lib.api.client import ApiClient
+from lib.api.models.dicom import GetDicom, GetDicomProcess, GetDicomProcesses, PostDicomProcesses
+
+
+def get_candidate_dicom(api: ApiClient, cand_id: int, visit_label: str):
+    response = api.get('v0.0.4-dev', f'candidates/{cand_id}/{visit_label}/dicoms')
+    return GetDicom.model_validate(response.json())
+
+
+def post_candidate_dicom(
+    api: ApiClient,
+    cand_id: int,
+    psc_id: str,
+    visit_label: str,
+    is_phantom: bool,
+    overwrite: bool,
+    file_path: str,
+):
+    data = {
+        'Json': json.dumps({
+            'CandID': cand_id,
+            'PSCID': psc_id,
+            'VisitLabel': visit_label,
+            'IsPhantom': is_phantom,
+            'Overwrite': overwrite,
+        }),
+    }
+
+    files = {
+        'File': (os.path.basename(file_path), open(file_path, 'rb'), 'application/x-tar'),
+    }
+
+    response = api.post('v0.0.4-dev', f'candidates/{cand_id}/{visit_label}/dicoms', data=data, files=files)
+    return response.headers['Location']
+
+
+def get_candidate_dicom_archive(api: ApiClient, cand_id: int, visit_label: str, tar_name: str):
+    api.get('v0.0.4-dev', f'candidates/{cand_id}/{visit_label}/dicoms/{tar_name}')
+    # TODO: Handle returned file
+
+
+def get_candidate_dicom_processes(api: ApiClient, cand_id: int, visit_label: str, tar_name: str):
+    response = api.get('v0.0.4-dev', f'candidates/{cand_id}/{visit_label}/dicoms/{tar_name}/processes')
+    return GetDicomProcesses.model_validate(response.json())
+
+
+def post_candidate_dicom_processes(api: ApiClient, cand_id: int, visit_label: str, tar_name: str, upload_id: int):
+    json = {
+        'ProcessType': 'mri_upload',
+        'MriUploadID': upload_id,
+    }
+
+    response = api.post(
+        'v0.0.4-dev',
+        f'/candidates/{cand_id}/{visit_label}/dicoms/{tar_name}/processes',
+        json=json,
+    )
+
+    return PostDicomProcesses.model_validate(response.json())
+
+
+def get_candidate_dicom_process(api: ApiClient, cand_id: int, visit_label: str, tar_name: str, process_id: int):
+    response = api.get('v0.0.4-dev', f'candidates/{cand_id}/{visit_label}/dicoms/{tar_name}/processes/{process_id}')
+    return GetDicomProcess.model_validate(response.json())

python/lib/api/models/dicom.py

@@ -0,0 +1,53 @@
+from typing import Literal, Optional
+
+from pydantic import BaseModel, Field
+
+
+class DicomArchiveSeries(BaseModel):
+    series_description : str                 = Field(alias='SeriesDescription')
+    series_number      : int                 = Field(alias='SeriesNumber')
+    echo_time          : Optional[str]       = Field(alias='EchoTime')
+    repetition_time    : Optional[str]       = Field(alias='RepetitionTime')
+    inversion_time     : Optional[str]       = Field(alias='InversionTime')
+    slice_thickness    : Optional[str]       = Field(alias='SliceThickness')
+    modality           : Literal['MR', 'PT'] = Field(alias='Modality')
+    series_uid         : str                 = Field(alias='SeriesUID')
+
+
+class DicomArchive(BaseModel):
+    tar_name     : str                         = Field(alias='Tarname')
+    patient_name : str                         = Field(alias='Patientname')
+    series       : list[DicomArchiveSeries] = Field(alias='SeriesInfo')
+
+
+class DicomMeta(BaseModel):
+    cand_id     : int = Field(alias='CandID')
+    visit_label : str = Field(alias='Visit')
+
+
+class GetDicom(BaseModel):
+    meta : DicomMeta          = Field(alias='Meta')
+    tars : list[DicomArchive] = Field(alias='DicomTars')
+
+
+class GetDicomProcess(BaseModel):
+    end_time  : Optional[str] = Field(alias='END_TIME')
+    exit_code : Optional[int] = Field(alias='EXIT_CODE')
+    id        : int           = Field(alias='ID')
+    pid       : int           = Field(alias='PID')
+    progress  : str           = Field(alias='PROGRESS')
+    state     : str           = Field(alias='STATE')
+
+
+class DicomUpload(BaseModel):
+    upload_id : int                   = Field(alias='MriUploadID')
+    processes : list[GetDicomProcess] = Field(alias='Processes')
+
+
+class PostDicomProcesses(BaseModel):
+    link      : str                   = Field(alias='Link')
+    processes : list[GetDicomProcess] = Field(alias='ProcessState')
+
+
+class GetDicomProcesses(BaseModel):
+    uploads : list[DicomUpload] = Field(alias='MriUploads')

python/requirements.txt

@@ -11,9 +11,11 @@ nose
 numpy
 protobuf>=3.0.0
 pybids==0.17.0
+pydantic
 pyright
 pytest
 python-dateutil
+requests
 ruff
 scikit-learn
 scipy