diff options
Diffstat (limited to 'gn3/llms/client.py')
| -rw-r--r-- | gn3/llms/client.py | 216 |
1 files changed, 61 insertions, 155 deletions
diff --git a/gn3/llms/client.py b/gn3/llms/client.py index 042becd..d57bca2 100644 --- a/gn3/llms/client.py +++ b/gn3/llms/client.py @@ -1,72 +1,60 @@ -# pylint: skip-file +"""Module Contains code for making request to fahamu Api""" +# pylint: disable=C0301 import json -import string -import os -import datetime import time -import requests -from requests import Session -from urllib.parse import urljoin -from requests.packages.urllib3.util.retry import Retry -from requests import HTTPError +import requests from requests import Session from requests.adapters import HTTPAdapter -from urllib.request import urlretrieve -from urllib.parse import quote -from gn3.llms.errors import UnprocessableEntity -from gn3.llms.errors import LLMError +from requests.adapters import Retry -basedir = os.path.join(os.path.dirname(__file__)) +from gn3.llms.errors import LLMError class TimeoutHTTPAdapter(HTTPAdapter): + """Set a default timeout for HTTP calls """ def __init__(self, timeout, *args, **kwargs): - """TimeoutHTTPAdapter constructor. - Args: - timeout (int): How many seconds to wait for the server to send data before - giving up. - """ + """TimeoutHTTPAdapter constructor.""" self.timeout = timeout super().__init__(*args, **kwargs) - def send(self, request, **kwargs): + def send(self, *args, **kwargs): """Override :obj:`HTTPAdapter` send method to add a default timeout.""" - timeout = kwargs.get("timeout") - if timeout is None: - kwargs["timeout"] = self.timeout - - return super().send(request, **kwargs) + kwargs["timeout"] = ( + kwargs["timeout"] if kwargs.get("timeout") else self.timeout + ) + return super().send(*args, **kwargs) class GeneNetworkQAClient(Session): """GeneNetworkQA Client This class provides a client object interface to the GeneNetworkQA API. - It extends the `requests.Session` class and includes authorization, base URL, + It extends the `requests.Session` class and includes authorization, + base URL, request timeouts, and request retries. Args: - account (str): Base address subdomain. api_key (str): API key. - version (str, optional): API version, defaults to "v3". timeout (int, optional): Timeout value, defaults to 5. total_retries (int, optional): Total retries value, defaults to 5. - backoff_factor (int, optional): Retry backoff factor value, defaults to 30. + backoff_factor (int, optional): Retry backoff factor value, + defaults to 30. Usage: from genenetworkqa import GeneNetworkQAClient - gnqa = GeneNetworkQAClient(account="account-name", api_key="XXXXXXXXXXXXXXXXXXX...") + gnqa = GeneNetworkQAClient(account="account-name", + api_key="XXXXXXXXXXXXXXXXXXX...") """ - BASE_URL = 'https://genenetwork.fahamuai.com/api/tasks' - - def __init__(self, account, api_key, version="v3", timeout=30, total_retries=5, backoff_factor=30): + def __init__(self, api_key, timeout=30, + total_retries=5, backoff_factor=30): super().__init__() self.headers.update( {"Authorization": "Bearer " + api_key}) - self.answer_url = f"{self.BASE_URL}/answers" - self.feedback_url = f"{self.BASE_URL}/feedback" + self.base_url = "https://genenetwork.fahamuai.com/api/tasks" + self.answer_url = f"{self.base_url}/answers" + self.feedback_url = f"{self.base_url}/feedback" adapter = TimeoutHTTPAdapter( timeout=timeout, @@ -80,141 +68,59 @@ class GeneNetworkQAClient(Session): self.mount("https://", adapter) self.mount("http://", adapter) - @staticmethod - def format_bibliography_info(bib_info): - - if isinstance(bib_info, str): - # Remove '.txt' - bib_info = bib_info.removesuffix('.txt') - elif isinstance(bib_info, dict): - # Format string bibliography information - bib_info = "{0}.{1}.{2}.{3} ".format(bib_info.get('author', ''), - bib_info.get('title', ''), - bib_info.get('year', ''), - bib_info.get('doi', '')) - return bib_info - - @staticmethod - def ask_the_documents(extend_url, my_auth): + def get_answer_using_task_id(self, extend_url, my_auth): + """call this method with task id to fetch response""" try: - response = requests.post( - base_url + extend_url, data={}, headers=my_auth) + response = requests.get( + self.answer_url + extend_url, data={}, headers=my_auth) response.raise_for_status() - except requests.exceptions.RequestException as e: - # Handle the exception appropriately, e.g., log the error - raise RuntimeError(f"Error making the request: {e}") - - if response.status_code != 200: - return negative_status_msg(response), 0 - - task_id = get_task_id_from_result(response) - response = get_answer_using_task_id(task_id, my_auth) - - if response.status_code != 200: - - return negative_status_msg(response), 0 - - return response, 1 + return response + except requests.exceptions.RequestException as error: + raise error @staticmethod def negative_status_msg(response): + """ handler for non 200 response from fahamu api""" return f"Error: Status code -{response.status_code}- Reason::{response.reason}" - # return f"Problems\n\tStatus code => {response.status_code}\n\tReason => {response.reason}" - def ask(self, exUrl, *args, **kwargs): - askUrl = self.BASE_URL + exUrl - res = self.custom_request('POST', askUrl, *args, **kwargs) - if (res.status_code != 200): - return self.negative_status_msg(res), 0 - task_id = self.getTaskIDFromResult(res) - return res, task_id + def ask(self, ex_url, *args, **kwargs): + """fahamu ask api interface""" + res = self.custom_request('POST', f"{self.base_url}{ex_url}", *args, **kwargs) + if res.status_code != 200: + return f"Error: Status code -{res.status_code}- Reason::{res.reason}", 0 + return res, json.loads(res.text) def get_answer(self, taskid, *args, **kwargs): - query = self.answer_url + self.extendTaskID(taskid) - res = self.custom_request('GET', query, *args, **kwargs) - if (res.status_code != 200): - return self.negative_status_msg(res), 0 - return res, 1 + """Fahamu get answer interface""" + try: + query = f"{self.answer_url}?task_id={taskid['task_id']}" + res = self.custom_request('GET', query, *args, **kwargs) + if res.status_code != 200: + return f"Error: Status code -{res.status_code}- Reason::{res.reason}", 0 + return res, 1 + except TimeoutError: + return "Timeout error occured:try to rephrase your query", 0 def custom_request(self, method, url, *args, **kwargs): - + """ make custom request to fahamu api ask and get response""" max_retries = 50 retry_delay = 3 - - for i in range(max_retries): + for _i in range(max_retries): try: response = super().request(method, url, *args, **kwargs) response.raise_for_status() - + if response.ok: + if method.lower() == "get" and response.json().get("data") is None: + time.sleep(retry_delay) + continue + return response + else: + time.sleep(retry_delay) except requests.exceptions.HTTPError as error: - if error.response.status_code ==500: - raise LLMError(error.request, error.response, f"Response Error,status_code:{error.response.status_code},Reason: Use of Invalid Token") - elif error.response.status_code ==404: - raise LLMError(error.request,error.response,f"404 Client Error: Not Found for url: {self.BASE_URL}") - raise error - + if error.response.status_code == 500: + raise LLMError(error.request, error.response, f"Response Error with:status_code:{error.response.status_code},Reason for error: Use of Invalid Fahamu Token") from error + raise LLMError(error.request, error.response, + f"HTTP error occurred with error status:{error.response.status_code}") from error except requests.exceptions.RequestException as error: - raise error - - - - - if response.ok: - if method.lower() == "get" and response.json().get("data") is None: - time.sleep(retry_delay) - continue - else: - return response - else: - time.sleep(retry_delay) - return response - - @staticmethod - def get_task_id_from_result(response): - task_id = json.loads(response.text) - result = f"?task_id={task_id.get('task_id', '')}" - return result - - @staticmethod - def get_answer_using_task_id(extend_url, my_auth): - try: - response = requests.get( - answer_url + extend_url, data={}, headers=my_auth) - response.raise_for_status() - return response - except requests.exceptions.RequestException as error: - # Handle the exception appropriately, e.g., log the error - raise error - - @staticmethod - def filter_response_text(val): - """ - Filters out non-printable characters from the input string and parses it as JSON. - - Args: - val (str): Input string to be filtered and parsed. - - Returns: - dict: Parsed JSON object. - # remove this - """ - return json.loads(''.join([str(char) for char in val if char in string.printable])) - - def getTaskIDFromResult(self, res): - return json.loads(res.text) - - def extendTaskID(self, task_id): - return '?task_id=' + str(task_id['task_id']) - - def get_gnqa(self, query): - qstr = quote(query) - res, task_id = api_client.ask('?ask=' + qstr) - res, success = api_client.get_answer(task_id) - - if success == 1: - resp_text = filter_response_text(res.text) - answer = resp_text.get('data', {}).get('answer', '') - context = resp_text.get('data', {}).get('context', '') - return answer, context - else: - return res, "Unfortunately, I have nothing." + raise error + raise TimeoutError |