File size: 7,259 Bytes
c49b21b |
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 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 |
"""
client.py – Base HTTP client for CoinDesk API.
This module provides the BaseClient class that handles HTTP requests
to the CoinDesk API with proper authentication and error handling.
"""
import requests
import json
from typing import Dict, Any, Optional
from urllib.parse import urljoin, urlencode
import config
class APIError(Exception):
"""Custom exception for API errors."""
def __init__(self, message: str, status_code: int = None, response: Any = None):
self.message = message
self.status_code = status_code
self.response = response
super().__init__(self.message)
class BaseClient:
"""
Base HTTP client for CoinDesk API requests.
Handles authentication, request formatting, and error handling.
"""
def __init__(self, base_url: str = None, headers: Dict[str, str] = None):
"""
Initialize the base client.
Args:
base_url: Base URL for the API (defaults to config.BASE_URL)
headers: Default headers (defaults to config.HEADERS)
"""
self.base_url = base_url or config.BASE_URL
self.headers = headers or config.HEADERS.copy()
self.session = requests.Session()
self.session.headers.update(self.headers)
def _make_request(self, method: str, endpoint: str, params: Dict[str, Any] = None,
data: Dict[str, Any] = None, **kwargs) -> Dict[str, Any]:
"""
Make an HTTP request to the API.
Args:
method: HTTP method (GET, POST, PUT, DELETE)
endpoint: API endpoint path
params: URL parameters
data: Request body data
**kwargs: Additional arguments for requests
Returns:
dict: JSON response from the API
Raises:
APIError: If the request fails or returns an error status
"""
# Construct full URL
url = urljoin(self.base_url, endpoint.lstrip('/'))
# Clean up parameters (remove None values)
if params:
params = {k: v for k, v in params.items() if v is not None}
try:
# Make the request
response = self.session.request(
method=method,
url=url,
params=params,
json=data,
**kwargs
)
# Log the request for debugging
print(f"[DEBUG] {method} {url}")
if params:
print(f"[DEBUG] Params: {params}")
print(f"[DEBUG] Status: {response.status_code}")
# Check if request was successful
if response.status_code == 200:
try:
return response.json()
except json.JSONDecodeError:
# If response is not JSON, return the text
return {"data": response.text, "status": "success"}
else:
# Handle different error status codes
error_message = f"API request failed with status {response.status_code}"
try:
error_data = response.json()
if 'error' in error_data:
error_message = error_data['error']
elif 'message' in error_data:
error_message = error_data['message']
except json.JSONDecodeError:
error_message = f"{error_message}: {response.text}"
raise APIError(
message=error_message,
status_code=response.status_code,
response=response
)
except requests.exceptions.RequestException as e:
raise APIError(f"Request failed: {str(e)}")
def get(self, endpoint: str, params: Dict[str, Any] = None, **kwargs) -> Dict[str, Any]:
"""
Make a GET request.
Args:
endpoint: API endpoint path
params: URL parameters
**kwargs: Additional arguments for requests
Returns:
dict: JSON response from the API
"""
return self._make_request('GET', endpoint, params=params, **kwargs)
def post(self, endpoint: str, data: Dict[str, Any] = None,
params: Dict[str, Any] = None, **kwargs) -> Dict[str, Any]:
"""
Make a POST request.
Args:
endpoint: API endpoint path
data: Request body data
params: URL parameters
**kwargs: Additional arguments for requests
Returns:
dict: JSON response from the API
"""
return self._make_request('POST', endpoint, params=params, data=data, **kwargs)
def put(self, endpoint: str, data: Dict[str, Any] = None,
params: Dict[str, Any] = None, **kwargs) -> Dict[str, Any]:
"""
Make a PUT request.
Args:
endpoint: API endpoint path
data: Request body data
params: URL parameters
**kwargs: Additional arguments for requests
Returns:
dict: JSON response from the API
"""
return self._make_request('PUT', endpoint, params=params, data=data, **kwargs)
def delete(self, endpoint: str, params: Dict[str, Any] = None, **kwargs) -> Dict[str, Any]:
"""
Make a DELETE request.
Args:
endpoint: API endpoint path
params: URL parameters
**kwargs: Additional arguments for requests
Returns:
dict: JSON response from the API
"""
return self._make_request('DELETE', endpoint, params=params, **kwargs)
def close(self):
"""Close the HTTP session."""
self.session.close()
def __enter__(self):
"""Context manager entry."""
return self
def __exit__(self, exc_type, exc_val, exc_tb):
"""Context manager exit."""
self.close()
# Convenience function to create a client instance
def create_client(base_url: str = None, headers: Dict[str, str] = None) -> BaseClient:
"""
Create a new BaseClient instance.
Args:
base_url: Base URL for the API
headers: Default headers
Returns:
BaseClient: Configured client instance
"""
return BaseClient(base_url=base_url, headers=headers)
# Test function to verify the client works
def test_client():
"""Test the base client functionality."""
try:
with create_client() as client:
# Test a simple endpoint (you might need to adjust this based on your API)
response = client.get("/index/cc/v1/markets")
print("Client test successful!")
print(f"Response keys: {list(response.keys()) if isinstance(response, dict) else 'Not a dict'}")
return True
except Exception as e:
print(f"Client test failed: {e}")
return False
if __name__ == "__main__":
test_client() |