| |
| from __future__ import annotations |
|
|
| import json |
| from typing import Any, Dict, Union |
| from urllib.parse import urljoin |
|
|
| import requests |
|
|
|
|
| def make_api_call( |
| host: Any = "", |
| endpoint: Any = "", |
| method: Any = "GET", |
| headers: Dict[str, str] = {"Content-Type": "application/json"}, |
| query_params: Union[Dict[str, Any], str, Any] = {}, |
| body: str = "", |
| timeout_secs: int = 60, |
| ) -> str: |
| """Make an API call to a given host and endpoint""" |
| response = {} |
| if not (isinstance(host, str) and isinstance(endpoint, str) and isinstance(method, str)): |
| raise ValueError("host, endpoint, method, and body must be a string") |
|
|
| allowed_methods = ["GET", "POST", "PUT", "DELETE"] |
| if method not in allowed_methods: |
| raise ValueError(f"method must be one of {allowed_methods}") |
|
|
| if not query_params: |
| query_params = {} |
| elif isinstance(query_params, str): |
| try: |
| query_params = json.loads(query_params) |
| except json.JSONDecodeError: |
| raise ValueError( |
| "query_params must be a dictionary or a JSON string", |
| ) |
| elif not isinstance(query_params, dict): |
| raise ValueError("query_params must be a dictionary or a JSON string") |
|
|
| if not host.startswith(("http://", "https://")): |
| normalized_host: str = f"https://{host}" |
| else: |
| normalized_host = host |
|
|
| url = urljoin(normalized_host, endpoint) |
|
|
| try: |
| if method not in allowed_methods: |
| raise ValueError(f"method must be one of {allowed_methods}") |
| response = requests.request(method=method, url=url, headers=headers, json=body, timeout=timeout_secs) |
|
|
| response_text = response.text |
| response = { |
| "status": "success", |
| "status_code": response.status_code, |
| "response": response_text, |
| } |
| except requests.exceptions.RequestException as e: |
| response = { |
| "status": "error", |
| "status_code": 500, |
| "response": str(e), |
| } |
| return json.dumps(response) |
|
|
