updated plugin to use auth token

Author: Sandip117

base_client.py

@@ -1,9 +1,6 @@
 from abc import ABC, abstractmethod
 
 class BaseClient(ABC):
-    @abstractmethod
-    def create_con(self, params: dict):
-        pass
 
     @abstractmethod
     def pacs_pull(self):

chrisClient.py

@@ -1,10 +1,8 @@
 ### Python Chris Client Implementation ###
 
 from base_client import BaseClient
-from chrisclient import client
-from chris_pacs_service import PACSClient
 import json
-import time
+import requests
 from loguru import logger
 import sys
 from pipeline import Pipeline
@@ -22,23 +20,29 @@
 logger.add(sys.stderr, format=logger_format)
 
 class ChrisClient(BaseClient):
-    def __init__(self, url: str, username: str, password: str):
-        self.cl = client.Client(url, username, password)
-        self.cl.pacs_series_url = f"{url}/pacs/series/"
-        self.req = PACSClient(self.cl.pacs_series_url,username,password)
-
-    def create_con(self,params:dict):
-        return self.cl
+    def __init__(self, url: str, token: str):
+        self.api_base = url.rstrip('/')
+        self.auth = token
+        self.headers = {"Content-Type": "application/json", "Authorization": f"Token {token}"}
+        self.pacs_series_url = f"{url}/pacs/series/"
 
     def health_check(self):
-        return self.cl.get_chris_instance()
+        endpoint = f"{self.api_base}/"
+        response = requests.request("GET", endpoint, headers=self.headers, timeout=30)
+
+        response.raise_for_status()
+
+        try:
+            return response.json()
+        except ValueError:
+            return response.text
 
     def pacs_pull(self):
         pass
     def pacs_push(self):
         pass
     def anonymize(self, params: dict, pv_id: int):
-        pipe = Pipeline(self.cl)
+        pipe = Pipeline(self.api_base, self.auth)
         plugin_params = {
             'PACS-query': {
                 "PACSurl": params["pull"]["url"],
@@ -52,9 +56,7 @@ def anonymize(self, params: dict, pv_id: int):
                 "copyInputFile": True
             },
             'verify-registration': {
-                "CUBEurl": self.cl.url,
-                "CUBEuser": self.cl.username,
-                "CUBEpassword": self.cl.password,
+                "CUBEurl": self.api_base,
                 "inputJSONfile": "search_results.json",
                 "folderName": params["push"]["Folder name"],
                 "neuroDicomLocation": params["push"]["Dicom path"],

chris_pacs_service.py

@@ -2,6 +2,9 @@
 from chrisclient import request
 from loguru import logger
 import sys
+from requests.exceptions import RequestException, Timeout, HTTPError
+from tenacity import retry, wait_exponential, stop_after_attempt, retry_if_exception_type
+from urllib.parse import urlencode
 
 LOG = logger.debug
 
@@ -18,19 +21,44 @@
 
 
 class PACSClient(object):
-    def __init__(self, url: str, username: str, password: str):
-        self.cl = request.Request(username, password)
+    def __init__(self, url: str, token: str):
+        self.api_base = url.rstrip('/')
+        self.auth = token
+        self.headers = {"Content-Type": "application/json"}
         self.pacs_series_search_url = f"{url}search/"
 
+    # --------------------------
+    # Retryable request handler
+    # --------------------------
+    @retry(
+        retry=retry_if_exception_type((RequestException, Timeout, HTTPError)),
+        wait=wait_exponential(multiplier=1, min=2, max=10),
+        stop=stop_after_attempt(5),
+        reraise=True
+    )
+    def make_request(self, method, endpoint, **kwargs):
+        response = requests.request(method, endpoint, headers=self.headers, auth=self.auth, timeout=30, **kwargs)
+        response.raise_for_status()
+
+        try:
+            return response.json()
+        except ValueError:
+            return response.text
+
 
     def get_pacs_files(self, params: dict):
+        """
+        Get PACS folder path
+        """
         l_dir_path = set()
-        resp = self.cl.get(self.pacs_series_search_url,params)
-        LOG(resp)
-        for item in resp.items:
-            for link in item.links:
-                folder = self.cl.get(link.href)
-                for item_folder in folder.items:
-                    path = item_folder.data.path.value
-                    l_dir_path.add(path)
+        query_string = urlencode(params)
+        response = self.make_request("GET", f"{self.pacs_series_search_url}?{query_string}")
+        for item in response.get("collection", {}).get("items", []):
+            for link in item.get("links", []):
+                folder = self.make_request("GET",link.get("href"))
+                for item_folder in folder.get("collection", {}).get("items", []):
+                    for field in item_folder.get("data", []):
+                        if field.get("name") == "path":
+                            l_dir_path.add(field.get("value"))
+
         return ','.join(l_dir_path)

dypxFlow.py

@@ -6,13 +6,10 @@
 from loguru import logger
 from chris_plugin import chris_plugin, PathMapper
 import pandas as pd
-import json
-import itertools
 from collections import ChainMap
 from chrisClient import ChrisClient
 import pfdcm
 import sys
-import time
 import os
 import concurrent.futures
 
@@ -29,7 +26,7 @@
 logger.remove()
 logger.add(sys.stderr, format=logger_format)
 
-__version__ = '1.0.7'
+__version__ = '1.0.8'
 
 DISPLAY_TITLE = r"""
        _           _                ______ _               
@@ -68,20 +65,15 @@
     help="CUBE URL. Please include api version in the url endpoint."
 )
 parser.add_argument(
-    "--CUBEuser",
-    default="chris",
-    help="CUBE/ChRIS username"
+    "--CUBEtoken",
+    default="",
+    help="CUBE/ChRIS user token"
 )
 parser.add_argument(
     "--maxThreads",
     default=4,
     help="max number of parallel threads"
 )
-parser.add_argument(
-    "--CUBEpassword",
-    default="chris1234",
-    help="CUBE/ChRIS password"
-)
 parser.add_argument(
     "--thread",
     help="use threading to branch in parallel",
@@ -97,7 +89,7 @@
     default=False,
 )
 parser.add_argument(
-    '--PACSurl',
+    '--PFDCMurl',
     default='',
     type=str,
     help='endpoint URL of pfdcm. Please include api version in the url endpoint.'
@@ -108,11 +100,6 @@
     type=str,
     help='name of the PACS'
 )
-parser.add_argument(
-    "--pftelDB",
-    help="an optional pftel telemetry logger, of form '<pftelURL>/api/v1/<object>/<collection>/<event>'",
-    default="",
-)
 
 def skip_condition(row):
     # Skip rows where starting column says 'no'
@@ -172,7 +159,7 @@ def main(options: Namespace, inputdir: Path, outputdir: Path):
                 results: Iterator = executor.map(lambda t: register_and_anonymize(options, t, options.wait), l_job)
 
             # Wait for all tasks to complete
-            # executor.shutdown(wait=True)
+            executor.shutdown(wait=True)
         else:
             for d_job in l_job:
                 response = register_and_anonymize(options, d_job)
@@ -200,12 +187,12 @@ def register_and_anonymize(options: Namespace, d_job: dict, wait: bool = False):
     """
     resp = {}
     d_job["pull"] = {
-        "url": options.PACSurl,
+        "url": options.PFDCMurl,
         "pacs": options.PACSname
     }
     LOG(d_job)
     if not d_job["push"]["status"]:
-        cube_con = ChrisClient(options.CUBEurl, options.CUBEuser, options.CUBEpassword)
+        cube_con = ChrisClient(options.CUBEurl, options.CUBEtoken)
         d_ret = cube_con.anonymize(d_job, options.pluginInstanceID)
     else:
         d_ret = d_job["push"]
@@ -226,14 +213,16 @@ def health_check(options) -> bool:
         return False
     try:
         # create connection object
-        cube_con = ChrisClient(options.CUBEurl, options.CUBEuser, options.CUBEpassword)
+        if not options.CUBEtoken:
+            options.CUBEtoken = os.environ['CHRIS_USER_TOKEN']
+        cube_con = ChrisClient(options.CUBEurl, options.CUBEtoken)
         cube_con.health_check()
     except Exception as ex:
         LOG(ex)
         return False
     try:
         # pfdcm health check
-        pfdcm.health_check(options.PACSurl)
+        pfdcm.health_check(options.PFDCMurl)
     except Exception as ex:
         LOG(ex)
         return False

pipeline.py

@@ -61,11 +61,9 @@ def compute_workflow_nodes_info(pipeline_default_parameters: list[dict], include
 
 
 class Pipeline:
-    def __init__(self, client):
-        self.cl = client
-        self.api_base = self.cl.url.rstrip('/')
-        self.auth = HTTPBasicAuth(self.cl.username, self.cl.password)
-        self.headers = {"Content-Type": "application/json"}
+    def __init__(self, url: str, token: str):
+        self.api_base = url.rstrip('/')
+        self.headers = {"Content-Type": "application/json", "Authorization": f"Token {token}"}
 
     # --------------------------
     # Retryable request handler
@@ -78,7 +76,7 @@ def __init__(self, client):
     )
     def make_request(self, method: str, endpoint: str, **kwargs):
         url = f"{self.api_base}{endpoint}"
-        response = requests.request(method, url, headers=self.headers, auth=self.auth, timeout=30, **kwargs)
+        response = requests.request(method, url, headers=self.headers, timeout=30, **kwargs)
         response.raise_for_status()
 
         try:
@@ -88,7 +86,7 @@ def make_request(self, method: str, endpoint: str, **kwargs):
 
     def post_request(self, endpoint: str, **kwargs):
         url = f"{self.api_base}{endpoint}"
-        response = requests.request("POST", url, headers=self.headers, auth=self.auth, timeout=30, **kwargs)
+        response = requests.request("POST", url, headers=self.headers, timeout=30, **kwargs)
         response.raise_for_status()
 
         try: