import json
import logging
import time
from dataclasses import field, dataclass
from enum import Enum
from typing import Dict, Any, Optional, Tuple
import requests
from olc.exception.exception import OlcBaseException
logger = logging.getLogger(__name__)
class HttpMethod(Enum):
GET = "GET"
POST = "POST"
PUT = "PUT"
DELETE = "DELETE"
@dataclass
class Request:
method: HttpMethod
url: str
headers: Dict[str, str] = field(default_factory=dict)
params: Dict[str, Any] = field(default_factory=dict)
data: Optional[Any] = None
json_data: Optional[Any] = None
def __repr__(self):
return f"<request: {self.method} {self.url}>"
def __str__(self):
return self.__repr__()
@dataclass
class Response:
status_code: int
url: str
headers: Dict[str, str]
content: Optional[bytes]
text: Optional[str]
elapsed: float
def json(self) -> Any:
try:
return json.loads(self.text)
except json.decoder.JSONDecodeError as e:
raise ValueError(f"response: {self.text} is not json") from e
def ok(self) -> bool:
return 200 <= self.status_code < 300
def __repr__(self):
return f"<response: {self.status_code} {self.url}>, elapsed: {self.elapsed}"
class HttpException(OlcBaseException):
def __init__(self, message: str, code: Optional[int] = None) -> None:
super().__init__(message, code)
def __repr__(self) -> str:
return f"HttpException(code={self.code}, message={self.message})"
def __str__(self) -> str:
return f"HttpException(code={self.code}, message={self.message})"
class RequestInterceptor:
@staticmethod
def before_request(request: Request) -> Request:
return request
@staticmethod
def after_response(response: Response) -> Response:
return response
def on_error(self, request: Request, exception: Exception) -> None:
pass
class LoggingInterceptor(RequestInterceptor):
def before_request(self, request: Request) -> Request:
logger.info("request: %s", request)
logger.debug(
"request: headers= %s, params=%s, data= %s, json_data=%s",
request.headers, request.params, request.data, request.json_data
)
return request
def after_response(self, response: Response) -> Response:
logger.info("response success: %s", response)
logger.debug(
"response: headers = %s, content = %s, text= %s",
response.headers, response.content, response.text
)
return response
def on_error(self, request: Request, exception: Exception) -> None:
logger.error(f"request error: {request}, exception: {exception}")
class HttpClient:
RETRY_STATUS_CODES = {429, 500, 502, 503, 504}
RETRY_METHODS = {"GET", "POST", "PUT", "DELETE"}
def __init__(
self,
base_url: Optional[str] = None,
timeout: Tuple[int, int] = (10, 30),
interceptors: Optional[list[RequestInterceptor]] = None,
max_retries: int = 3,
):
self.base_url = base_url
self.timeout = timeout
self.max_retries = max_retries
self.interceptors = interceptors if interceptors is not None else [LoggingInterceptor()]
logger.info("httpclient init success: base_url: %s", self.base_url)
def _build_url(self, url: str) -> str:
if self.base_url and not url.startswith(("http://", "https://")):
return f"{self.base_url.rstrip('/')}/{url.lstrip('/')}"
return url
def _apply_interceptor_before(self, request: Request) -> Request:
for interceptor in self.interceptors:
try:
request = interceptor.before_request(request)
except Exception as e:
logger.warning("request pre-processing failed: %s", e)
return request
def _apply_interceptor_after(self, response: Response) -> Response:
for interceptor in self.interceptors:
try:
response = interceptor.after_response(response)
except Exception as e:
logger.warning("response post-processing failed: %s", e)
return response
def _apply_interceptor_on_error(self, request: Request, exception: Exception) -> None:
for interceptor in self.interceptors:
try:
interceptor.on_error(request, exception)
except Exception as e:
logger.warning("error interceptor failed: %s", e)
def add_interceptor(self, interceptor: RequestInterceptor):
if interceptor is not None:
self.interceptors.append(interceptor)
def request(self, req: Request) -> Response:
req = Request(req.method, self._build_url(req.url), req.headers, req.params, req.data, req.json_data)
try:
req = self._apply_interceptor_before(req)
except Exception as e:
logger.warning("request pre-processing failed: %s", e)
retry_count = 0
last_exception = None
should_retry = req.method.value in self.RETRY_METHODS
while retry_count <= self.max_retries:
try:
start_time = time.time()
response = requests.request(
method=req.method.value,
url=req.url,
headers=req.headers,
params=req.params,
data=req.data,
json=req.json_data,
timeout=self.timeout,
)
elapsed = time.time() - start_time
response_obj = Response(
status_code=response.status_code,
url=response.url,
headers=dict(response.headers),
content=response.content,
text=response.text,
elapsed=elapsed,
)
if should_retry and response.status_code in self.RETRY_STATUS_CODES and retry_count < self.max_retries:
retry_count += 1
wait_time = 2 ** retry_count
logger.warning(
f"Retrying request {retry_count}/{self.max_retries}, "
f"status code: {response.status_code}, "
f"waiting {wait_time}s"
)
time.sleep(wait_time)
continue
try:
response_obj = self._apply_interceptor_after(response_obj)
except Exception as e:
logger.warning("response post-processing failed: %s", e)
return response_obj
except requests.exceptions.RequestException as e:
last_exception = e
self._apply_interceptor_on_error(req, e)
break
except Exception as e:
last_exception = e
self._apply_interceptor_on_error(req, e)
break
if last_exception:
raise HttpException(f"request failed: {last_exception}")
def get(
self,
url: str,
params: Optional[Dict[str, Any]] = None,
headers: Optional[Dict[str, str]] = None,
) -> Response:
return self.request(
Request(HttpMethod.GET, url, headers=headers, params=params)
)
def post(
self,
url: str,
data: Optional[Any] = None,
json_data: Optional[Any] = None,
headers: Optional[Dict[str, str]] = None,
params: Optional[Dict[str, Any]] = None,
) -> Response:
return self.request(
Request(HttpMethod.POST, url, headers=headers, params=params, data=data, json_data=json_data)
)
def put(
self,
url: str,
data: Optional[Any] = None,
json_data: Optional[Any] = None,
headers: Optional[Dict[str, str]] = None,
params: Optional[Dict[str, Any]] = None,
) -> Response:
return self.request(
Request(HttpMethod.PUT, url, headers=headers, params=params, data=data, json_data=json_data)
)
def delete(
self,
url: str,
params: Optional[Dict[str, Any]] = None,
headers: Optional[Dict[str, str]] = None,
) -> Response:
return self.request(
Request(HttpMethod.DELETE, url, headers=headers, params=params)
)
def get_json(
self,
url: str,
params: Optional[Dict[str, Any]] = None,
headers: Optional[Dict[str, Any]] = None,
) -> Any:
response = self.get(url, params=params, headers=headers)
if not response.ok():
raise HttpException(f"get json failed: {response}", code=response.status_code)
try:
return response.json()
except ValueError as e:
raise HttpException(f"response is not valid json: {e}", code=response.status_code) from e
def post_json(
self,
url: str,
json_data: Optional[Any] = None,
headers: Optional[Dict[str, str]] = None,
params: Optional[Dict[str, Any]] = None,
) -> Any:
response = self.post(url, json_data=json_data, headers=headers, params=params)
if not response.ok():
raise HttpException(f"post json failed: {response}", code=response.status_code)
try:
return response.json()
except ValueError as e:
raise HttpException(f"response is not valid json: {e}", code=response.status_code) from e