Skip to content

Commit

Permalink
basic linting
Browse files Browse the repository at this point in the history
  • Loading branch information
Jermiah committed Nov 18, 2023
1 parent f1c2364 commit 8525b23
Show file tree
Hide file tree
Showing 3 changed files with 29 additions and 18 deletions.
25 changes: 13 additions & 12 deletions src/auth.py
Original file line number Diff line number Diff line change
Expand Up @@ -10,7 +10,6 @@ def __init__(self,
self.password = password
self.access_token = None
self.api_headers = None
# self.expiry_time = None

#username=nbia_guest&password=&client_id=NBIA&grant_type=password
def getToken(self):
Expand All @@ -29,27 +28,29 @@ def getToken(self):
}
token_url = 'https://services.cancerimagingarchive.net/nbia-api/oauth/token'
# Make a POST request to the token endpoint
response = requests.post(token_url, data=data) # Disable SSL verification for simplicity
response = requests.post(token_url, data=data)
# response.raise_for_status()

# Check if the request was successful
if response.status_code == 200:
token_data = response.json()
self.access_token = token_data.get('access_token') # save access token to self for later use
# save api headers to self for later use

self.access_token = token_data.get('access_token')
# save access token to self for later use
# self.api_headers = {
# 'Authorization': f'Bearer {self.access_token}',
# 'Accept': 'application/json'
# }
# TODO::implement refresh token functionality
self.expiry_time = time.ctime(time.time() + token_data.get('expires_in') )
self.expiry_time = time.ctime(
time.time() + token_data.get('expires_in') )
self.refresh_token = token_data.get('refresh_token')
self.refresh_expiry = token_data.get('refresh_expires_in')
self.scope = token_data.get('scope')
return self.access_token
else:
print(f"Failed to get access token. Status code: {response.status_code}")
print(f"Failed to get access token. Status code: \
{response.status_code}")

self.access_token = response.status_code
return response.status_code

## Getting this warning:
# InsecureRequestWarning: Unverified HTTPS request is being made to host
# 'services.cancerimagingarchive.net'. Adding certificate verification is
# strongly advised. See: https://urllib3.readthedocs.io/en/1.26.x/advanced-usage.html#ssl-warnings
#warnings.warn(
1 change: 0 additions & 1 deletion src/main.py

This file was deleted.

21 changes: 16 additions & 5 deletions src/nbia.py
Original file line number Diff line number Diff line change
Expand Up @@ -11,7 +11,8 @@ def __init__(self,
) -> None:

# Setup logger
self.logger = setup_logger(name = "NBIAClient", console_logging=True, log_level=log_level)
self.logger = setup_logger(
name = "NBIAClient", console_logging=True, log_level=log_level)

# Setup OAuth2 client
self.logger.info("Setting up OAuth2 client... with username %s", username)
Expand Down Expand Up @@ -67,7 +68,10 @@ def getCollectionPatientCount(self) -> list:
response = self.query_api(NBIA_ENDPOINTS.GET_COLLECTION_PATIENT_COUNT)
patientCount = []
for collection in response:
patientCount.append({"Collection": collection["criteria"], "PatientCount": collection["count"]})
patientCount.append({
"Collection": collection["criteria"],
"PatientCount": collection["count"]})

return patientCount

def getBodyPartCounts(self, collection: str = "", modality: str = "") -> list:
Expand All @@ -76,18 +80,25 @@ def getBodyPartCounts(self, collection: str = "", modality: str = "") -> list:
PARAMS["Collection"] = collection
if modality:
PARAMS["Modality"] = modality
response = self.query_api(endpoint = NBIA_ENDPOINTS.GET_BODY_PART_PATIENT_COUNT, params = PARAMS)
response = self.query_api(
endpoint = NBIA_ENDPOINTS.GET_BODY_PART_PATIENT_COUNT,
params = PARAMS)

bodyparts=[]
for bodypart in response:
bodyparts.append({"BodyPartExamined": bodypart["criteria"], "Count": bodypart["count"]})
bodyparts.append({
"BodyPartExamined": bodypart["criteria"],
"Count": bodypart["count"]})
return bodyparts

def getPatients(self, collection: str, modality: str) -> list:
assert collection is not None
assert modality is not None

PARAMS = {"Collection": collection,"Modality": modality}
response = self.query_api(endpoint = NBIA_ENDPOINTS.GET_PATIENT_BY_COLLECTION_AND_MODALITY, params = PARAMS)
response = self.query_api(
endpoint = NBIA_ENDPOINTS.GET_PATIENT_BY_COLLECTION_AND_MODALITY,
params = PARAMS)
patientList = [_["PatientId"] for _ in response]
return patientList

0 comments on commit 8525b23

Please sign in to comment.