Source code for kripodb.webservice.client

# Copyright 2016 Netherlands eScience Center
#
# Licensed under the Apache License, Version 2.0 (the 'License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#   http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
"""Module for Client for kripo web service"""
from __future__ import absolute_import
import requests
from rdkit.Chem.AllChem import MolFromMolBlock
from requests import HTTPError


[docs]class Incomplete(Exception): def __init__(self, message, absent_identifiers): super(Incomplete, self).__init__(message) self.absent_identifiers = absent_identifiers
[docs]class IncompleteFragments(Incomplete): def __init__(self, absent_identifiers, fragments): """List of fragments and list of identifiers for which no information could be found Args: absent_identifiers (List[str]): List of identifiers for which no information could be found fragments (List[dict]): List of fragment information that could be retrieved """ message = 'Some identifiers could not be found' super(IncompleteFragments, self).__init__(message, absent_identifiers) self.fragments = fragments
[docs]class IncompletePharmacophores(Incomplete): def __init__(self, absent_identifiers, pharmacophores): """List of fragments and list of identifiers for which no information could be found Args: absent_identifiers (List[str]): List of identifiers for which no information could be found pharmacophores (List[dict]): List of pharmacophores that could be retrieved """ message = 'Some identifiers could not be found' super(IncompletePharmacophores, self).__init__(message, absent_identifiers) self.pharmacophores = pharmacophores
[docs]class WebserviceClient(object): """Client for kripo web service Example: >>> client = WebserviceClient('http://localhost:8084/kripo') >>> client.similar_fragments('3j7u_NDP_frag24', 0.85) [{'query_frag_id': '3j7u_NDP_frag24', 'hit_frag_id': '3j7u_NDP_frag23', 'score': 0.8991}] Args: base_url (str): Base url of web service. e.g. http://localhost:8084/kripo """ def __init__(self, base_url): self.base_url = base_url
[docs] def similar_fragments(self, fragment_id, cutoff, limit=1000): """Find similar fragments to query. Args: fragment_id (str): Query fragment identifier cutoff (float): Cutoff, similarity scores below cutoff are discarded. limit (int): Maximum number of hits. Default is None for no limit. Returns: list[dict]: Query fragment identifier, hit fragment identifier and similarity score Raises: request.HTTPError: When fragment_id could not be found """ url = self.base_url + '/fragments/{fragment_id}/similar'.format(fragment_id=fragment_id) params = {'cutoff': cutoff, 'limit': limit} response = requests.get(url, params) response.raise_for_status() return response.json()
[docs] def fragments_by_pdb_codes(self, pdb_codes, chunk_size=450): """Retrieve fragments by their PDB code Args: pdb_codes (List[str]): List of PDB codes chunk_size (int): Number of PDB codes to retrieve in a single http request Returns: list[dict]: List of fragment information Raises: requests.HTTPError: When one of the PDB codes could not be found. """ return self._fetch_chunked_fragments('pdb_codes', pdb_codes, chunk_size)
[docs] def fragments_by_id(self, fragment_ids, chunk_size=100): """Retrieve fragments by their identifier Args: fragment_ids (List[str]): List of fragment identifiers chunk_size (int): Number of fragment to retrieve in a single http request Returns: list[dict]: List of fragment information Raises: IncompleteFragments: When one or more of the identifiers could not be found. """ return self._fetch_chunked_fragments('fragment_ids', fragment_ids, chunk_size)
def _fetch_chunked_fragments(self, idtype, ids, chunk_size): fragments = [] absent_identifiers = [] for start in range(0, len(ids), chunk_size): stop = chunk_size + start (chunk_fragments, chunk_absent_identifiers) = self._fetch_fragments(idtype, ids[start:stop]) fragments += chunk_fragments absent_identifiers += chunk_absent_identifiers if chunk_absent_identifiers: raise IncompleteFragments(absent_identifiers, fragments) return fragments def _fetch_fragments(self, idtype, ids): url = self.base_url + '/fragments?{idtype}={ids}'.format(idtype=idtype, ids=','.join(ids)) absent_identifiers = [] try: response = requests.get(url) response.raise_for_status() fragments = response.json() except HTTPError as e: if e.response.status_code == 404: body = e.response.json() fragments = body['fragments'] absent_identifiers = body['absent_identifiers'] else: raise e # Convert molblock string to RDKit Mol object for fragment in fragments: if fragment['mol'] is not None: fragment['mol'] = MolFromMolBlock(fragment['mol']) return fragments, absent_identifiers def pharmacophores(self, fragment_ids): absent_identifiers = [] pharmacophores = [] for fragment_id in fragment_ids: url = self.base_url + '/fragments/{0}.phar'.format(fragment_id) try: response = requests.get(url) response.raise_for_status() pharmacophore = response.text pharmacophores.append(pharmacophore) except HTTPError as e: if e.response.status_code == 404: pharmacophores.append(None) absent_identifiers.append(fragment_id) else: raise e if absent_identifiers: raise IncompletePharmacophores(absent_identifiers, pharmacophores) return pharmacophores