File size: 2,163 Bytes
3d3d712 |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 |
# This is used to define common functions/tools that could be used by different plugins
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)
|