File size: 1,859 Bytes
a51a15b |
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 |
import os
import requests
from typing import Dict, Any, Optional, TypedDict, Literal
class EndpointSchema(TypedDict):
route: str
method: Literal['GET', 'POST']
name: str
description: str
payload: Dict[str, Any]
class RapidDataProviderBase:
def __init__(self, base_url: str, endpoints: Dict[str, EndpointSchema]):
self.base_url = base_url
self.endpoints = endpoints
def get_endpoints(self):
return self.endpoints
def call_endpoint(
self,
route: str,
payload: Optional[Dict[str, Any]] = None
):
"""
Call an API endpoint with the given parameters and data.
Args:
endpoint (EndpointSchema): The endpoint configuration dictionary
params (dict, optional): Query parameters for GET requests
payload (dict, optional): JSON payload for POST requests
Returns:
dict: The JSON response from the API
"""
if route.startswith("/"):
route = route[1:]
endpoint = self.endpoints.get(route)
if not endpoint:
raise ValueError(f"Endpoint {route} not found")
url = f"{self.base_url}{endpoint['route']}"
headers = {
"x-rapidapi-key": os.getenv("RAPID_API_KEY"),
"x-rapidapi-host": url.split("//")[1].split("/")[0],
"Content-Type": "application/json"
}
method = endpoint.get('method', 'GET').upper()
if method == 'GET':
response = requests.get(url, params=payload, headers=headers)
elif method == 'POST':
response = requests.post(url, json=payload, headers=headers)
else:
raise ValueError(f"Unsupported HTTP method: {method}")
return response.json()
|