Skip to content
Open
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
69 changes: 37 additions & 32 deletions ctxpy/base.py
Original file line number Diff line number Diff line change
@@ -1,34 +1,34 @@
import json
import warnings
from pathlib import Path
from typing import Optional, Union, Iterable
from typing import Iterable, Optional, Union
from urllib.parse import quote

import pandas as pd
import requests
from urllib.parse import quote

from .utils import read_env, chunker
from .utils import chunker, read_env


class CTXConnection:
"""
Connection that passes API key and other variables needed for GET and POST calls to
API server.

Parameters
----------
x_api_key : str or None, default None
A user's API key provided to them for accessing CCTE's APIs. Will use value
provided in .env file if no key is provided.
env_path : str or None, default None
The .env file location. Will default to a user's home directory if no value is
The .env file location. Will default to a user's home directory if no value is
provided.

Attributes
----------
headers : dict
A dictionary of information used to make an API call

Methods
-------
get
Expand All @@ -37,38 +37,39 @@ class CTXConnection:
send data to resource and then retrieve information based on that data

"""

def __init__(
self,
x_api_key: Optional[str] = None,
env_path: Optional[Union[str, Path]] = None,
):
if isinstance(x_api_key, str):
self.host = "https://api-ccte.epa.gov/"
self.host = "https://comptox.epa.gov/ctx-api/"
self.headers = {"accept": "application/json", "x-api-key": x_api_key}

else:
config = read_env()
self.host = config["ctx_api_host"]
self.headers = {"accept":config['ctx_api_accept'],
"x-api-key":config['ctx_api_x_api_key']}
self.headers = {
"accept": config["ctx_api_accept"],
"x-api-key": config["ctx_api_x_api_key"],
}

def get(self, suffix: str):
"""
Request informaiton via API call

Paramters
---------
suffix : string
the suffix of the API call that will determine what is searched for and how

Returns
-------
dict, JSON information that was requested in the API call
"""
try:
self.response = requests.get(
f"{self.host}{suffix}", headers=self.headers
)
self.response = requests.get(f"{self.host}{suffix}", headers=self.headers)
except Exception as e:
## TODO: make this a more informative error message
raise SystemError(e)
Expand All @@ -84,7 +85,7 @@ def post(self, suffix: str, word: str):
"""
Request information via API call, but also supply stipulations on subsets or
specific aspects of returned information

Paramters
---------
suffix : string
Expand Down Expand Up @@ -113,8 +114,10 @@ def post(self, suffix: str, word: str):
raise SystemError(e)

return info

def batch(self, suffix: str, word: Iterable[str], batch_size: int, bracketed:bool = False):

def batch(
self, suffix: str, word: Iterable[str], batch_size: int, bracketed: bool = False
):
"""
There are some inconsistencies in how to provide 'batch' data to the API.
Sometimes the list of identifiers needs to be a bracketed list, other times it
Expand All @@ -127,7 +130,7 @@ def batch(self, suffix: str, word: Iterable[str], batch_size: int, bracketed:boo

chunks = []
for chunk in chunker(word, batch_size):

if bracketed:
## '["DTXSID001", "DTXSID002"]'
words = [quote(w, safe="") for w in chunk]
Expand All @@ -138,50 +141,52 @@ def batch(self, suffix: str, word: Iterable[str], batch_size: int, bracketed:boo

## TODO: Check that suffix is re-written on each loop,
## not appended to
chunks.extend(self.post(suffix = suffix, word=words))
chunks.extend(self.post(suffix=suffix, word=words))
return chunks



class HCDConnection:
"""
Connection to the Hazard Comparison Dashboard APIs, no API key is needed, only GET
Connection to the Hazard Comparison Dashboard APIs, no API key is needed, only GET
calls are allowed to the API server.





Methods
-------
get
request information from specified source using information in header

"""

def __init__(self):

self.host = "https://hcd.rtpnc.epa.gov/api/"
self.descriptors = "descriptors?type=toxprints&smiles="
self.headers = "&headers=TRUE"

def get(self, smiles: str):
"""
Request informaiton via API call

Paramters
---------
smiles : string
the suffix of the API call that will determine what is searched for and how

Returns
-------
dict, JSON information that was requested in the API call
"""
word = quote(smiles, safe="")
try:
with warnings.catch_warnings():
warnings.filterwarnings("ignore",category=requests.urllib3.connectionpool.InsecureRequestWarning)
warnings.filterwarnings(
"ignore",
category=requests.urllib3.connectionpool.InsecureRequestWarning,
)
self.response = requests.get(
f"{self.host}{self.descriptors}{word}{self.headers}",
verify = False
f"{self.host}{self.descriptors}{word}{self.headers}", verify=False
)
except Exception as e:
## TODO: make this a more informative error message
Expand All @@ -192,9 +197,9 @@ def get(self, smiles: str):
except json.JSONDecodeError as e:
## TODO: make this a more informative error message
raise SystemError(e)
if 'descriptors' not in info['chemicals'][0].keys():
if "descriptors" not in info["chemicals"][0].keys():
info = None
else:
info = info['chemicals'][0]['descriptors']
info = info["chemicals"][0]["descriptors"]

return info