API Reference - API Client

pyepisuite.api_client

EpiSuiteAPIClient

Source code in src/pyepisuite/api_client.py

class EpiSuiteAPIClient:
    def __init__(self, base_url=None, api_key=None):
        mode = os.getenv('PYEPISUITE_MODE', 'auto').strip().lower()
        if mode not in {'auto', 'local', 'remote'}:
            raise ValueError("PYEPISUITE_MODE must be one of: auto, local, remote")

        resolved_base_url = base_url
        self.local_mode = False

        if resolved_base_url is None:
            explicit_local_base = os.getenv('PYEPISUITE_LOCAL_BASE_URL')

            if mode == 'remote':
                resolved_base_url = DEFAULT_REMOTE_BASE_URL
            elif explicit_local_base:
                resolved_base_url = _with_api_suffix(explicit_local_base)
                self.local_mode = True
            elif mode == 'local' or _LocalRuntimeManager.has_local_assets():
                resolved_base_url = _with_api_suffix(_LocalRuntimeManager.ensure_started())
                self.local_mode = True
            else:
                resolved_base_url = DEFAULT_REMOTE_BASE_URL
        elif re.match(r'^https?://(127\.0\.0\.1|localhost)(:\d+)?(/.*)?$', resolved_base_url):
            self.local_mode = True

        self.base_url = resolved_base_url
        self.api_key = api_key

    def _headers(self):
        headers = {}
        if self.api_key:
            headers['Authorization'] = f'Bearer {self.api_key}'
        return headers

    def _parse_json_response(self, response, operation_name):
        response.raise_for_status()
        content_type = response.headers.get('content-type', '').lower()
        if 'application/json' not in content_type:
            snippet = response.text[:600].strip()
            raise ValueError(
                f'{operation_name} expected JSON but received {content_type or "unknown content type"}. '
                f'Response snippet: {snippet}'
            )
        return response.json()

    def search(self, query_term, time_out=10):
        """
        Search the EPISuite API with a query term (SMILES, CAS, or chemical name).

        Parameters:
            query_term (str): The term to search for.
            time_out (int): The time out for the request.

        Returns:
            List[Chemical]: A list of Chemical instances.
        """
        url = f'{self.base_url}/search'
        params = {'query': query_term}
        response = requests.get(url, params=params, headers=self._headers(), timeout=time_out)
        data = self._parse_json_response(response, 'search')

        # Convert each dictionary in the response to a Chemical instance
        ids = [Identifiers(**item) for item in data]
        return ids

    def submit(self, cas="", smiles=""):
        """
        Submit a CAS number or SMILES string to the EPISuite API.

        Parameters:
            cas (str): The CAS number of the chemical.
            smiles (str): The SMILES string of the chemical.

        Returns:
            dict: The JSON response from the API.

        Raises:
            ValueError: If neither 'cas' nor 'smiles' is provided.
        """
        if not cas and not smiles:
            raise ValueError("Either 'cas' or 'smiles' must be provided.")

        url = f'{self.base_url}/submit'
        params = {}
        if cas:
            params['cas'] = cas
        else:
            params['smiles'] = smiles

        response = requests.get(url, params=params, headers=self._headers(), timeout=120)
        try:
            return self._parse_json_response(response, 'submit')
        except ValueError as exc:
            if cas and self.local_mode and _is_cas_formatted(cas):
                normalized = _normalize_cas_for_local(cas)
                if normalized != cas:
                    retry_params = dict(params)
                    retry_params['cas'] = normalized
                    retry_response = requests.get(url, params=retry_params, headers=self._headers(), timeout=120)
                    return self._parse_json_response(retry_response, 'submit')
            raise exc

    @staticmethod
    def stop_local_runtime():
        """Stop the managed local runtime launched by this package."""
        stop_local_episuite_server()

search(query_term, time_out=10)

Search the EPISuite API with a query term (SMILES, CAS, or chemical name).

Parameters:

Name	Type	Description	Default
query_term	str	The term to search for.	required
time_out	int	The time out for the request.	10

Returns:

Type	Description
	List[Chemical]: A list of Chemical instances.

Source code in src/pyepisuite/api_client.py

def search(self, query_term, time_out=10):
    """
    Search the EPISuite API with a query term (SMILES, CAS, or chemical name).

    Parameters:
        query_term (str): The term to search for.
        time_out (int): The time out for the request.

    Returns:
        List[Chemical]: A list of Chemical instances.
    """
    url = f'{self.base_url}/search'
    params = {'query': query_term}
    response = requests.get(url, params=params, headers=self._headers(), timeout=time_out)
    data = self._parse_json_response(response, 'search')

    # Convert each dictionary in the response to a Chemical instance
    ids = [Identifiers(**item) for item in data]
    return ids

stop_local_runtime() staticmethod

Stop the managed local runtime launched by this package.

Source code in src/pyepisuite/api_client.py

@staticmethod
def stop_local_runtime():
    """Stop the managed local runtime launched by this package."""
    stop_local_episuite_server()

submit(cas='', smiles='')

Submit a CAS number or SMILES string to the EPISuite API.

Parameters:

Name	Type	Description	Default
cas	str	The CAS number of the chemical.	''
smiles	str	The SMILES string of the chemical.	''

Returns:

Name	Type	Description
dict		The JSON response from the API.

Raises:

Type	Description
ValueError	If neither 'cas' nor 'smiles' is provided.

Source code in src/pyepisuite/api_client.py

def submit(self, cas="", smiles=""):
    """
    Submit a CAS number or SMILES string to the EPISuite API.

    Parameters:
        cas (str): The CAS number of the chemical.
        smiles (str): The SMILES string of the chemical.

    Returns:
        dict: The JSON response from the API.

    Raises:
        ValueError: If neither 'cas' nor 'smiles' is provided.
    """
    if not cas and not smiles:
        raise ValueError("Either 'cas' or 'smiles' must be provided.")

    url = f'{self.base_url}/submit'
    params = {}
    if cas:
        params['cas'] = cas
    else:
        params['smiles'] = smiles

    response = requests.get(url, params=params, headers=self._headers(), timeout=120)
    try:
        return self._parse_json_response(response, 'submit')
    except ValueError as exc:
        if cas and self.local_mode and _is_cas_formatted(cas):
            normalized = _normalize_cas_for_local(cas)
            if normalized != cas:
                retry_params = dict(params)
                retry_params['cas'] = normalized
                retry_response = requests.get(url, params=retry_params, headers=self._headers(), timeout=120)
                return self._parse_json_response(retry_response, 'submit')
        raise exc

LocalEpiSuiteAPIClient

Bases: EpiSuiteAPIClient

Explicit local client that always targets the local EpiSuite runtime.

Source code in src/pyepisuite/api_client.py

class LocalEpiSuiteAPIClient(EpiSuiteAPIClient):
    """Explicit local client that always targets the local EpiSuite runtime."""

    def __init__(self, base_url=None, api_key=None):
        resolved_base_url = base_url or os.getenv('PYEPISUITE_LOCAL_BASE_URL')
        if resolved_base_url is None:
            resolved_base_url = _with_api_suffix(_LocalRuntimeManager.ensure_started())
        else:
            resolved_base_url = _with_api_suffix(resolved_base_url)

        super().__init__(base_url=resolved_base_url, api_key=api_key)
        self.local_mode = True

jar_is_supported(jar_path)

Whether this jar is new enough to serve the HTTP API this client speaks.

Source code in src/pyepisuite/api_client.py

def jar_is_supported(jar_path) -> bool:
    """Whether this jar is new enough to serve the HTTP API this client speaks."""
    version = _parse_version(read_jar_version(jar_path))
    return version is not None and version >= MINIMUM_LOCAL_JAR_VERSION

read_jar_version(jar_path)

Read Implementation-Version from a jar's manifest.

Cheap by design: it reads the zip entry rather than starting a JVM. Returns None for a jar that declares no version, which is how the legacy EpiSuiteCLI builds present themselves.

Source code in src/pyepisuite/api_client.py

def read_jar_version(jar_path) -> Optional[str]:
    """Read Implementation-Version from a jar's manifest.

    Cheap by design: it reads the zip entry rather than starting a JVM. Returns
    None for a jar that declares no version, which is how the legacy
    EpiSuiteCLI builds present themselves.
    """
    try:
        with zipfile.ZipFile(jar_path) as archive:
            manifest = archive.read('META-INF/MANIFEST.MF').decode('utf-8', 'replace')
    except (OSError, KeyError, zipfile.BadZipFile):
        return None
    # Manifest lines may be wrapped by a leading space; unwrap before matching.
    manifest = manifest.replace('\r\n', '\n').replace('\r', '\n').replace('\n ', '')
    match = re.search(r'^Implementation-Version:\s*(.+)$', manifest, re.M)
    return match.group(1).strip() if match else None

stop_local_episuite_server()

Stop the managed local EpiSuite runtime if it is running.

Source code in src/pyepisuite/api_client.py

def stop_local_episuite_server():
    """Stop the managed local EpiSuite runtime if it is running."""
    _LocalRuntimeManager.stop()