| import requests | |
| from config import API_URL | |
| _TIMEOUT_FAST = 10 | |
| _TIMEOUT_QUERY = 60 | |
| _TIMEOUT_UPLOAD = 120 | |
| _TIMEOUT_DATASET = 600 | |
| def health_check() -> dict: | |
| resp = requests.get(f"{API_URL}/health", timeout=_TIMEOUT_FAST) | |
| resp.raise_for_status() | |
| return resp.json() | |
| def upload_pdfs(files) -> dict: | |
| payload = [("files", (f.name, f.read(), "application/pdf")) for f in files] | |
| resp = requests.post(f"{API_URL}/upload_pdfs/", files=payload, timeout=_TIMEOUT_UPLOAD) | |
| resp.raise_for_status() | |
| return resp.json() | |
| def ask_question(question: str) -> dict: | |
| resp = requests.post(f"{API_URL}/ask/", data={"question": question}, timeout=_TIMEOUT_QUERY) | |
| resp.raise_for_status() | |
| return resp.json() | |
| def get_available_datasets() -> list[dict]: | |
| resp = requests.get(f"{API_URL}/hf_datasets/", timeout=_TIMEOUT_FAST) | |
| resp.raise_for_status() | |
| return resp.json().get("available_datasets", []) | |
| def load_datasets(dataset_names: list[str], max_samples: int = 500) -> dict: | |
| resp = requests.post( | |
| f"{API_URL}/hf_datasets/load/", | |
| json={"datasets": dataset_names, "max_samples": max_samples}, | |
| timeout=_TIMEOUT_DATASET, | |
| ) | |
| resp.raise_for_status() | |
| return resp.json() | |