libretime/api-client/libretime_api_client/_client.py

95 lines
2.4 KiB
Python
Raw Permalink Normal View History

2023-02-26 01:27:00 +01:00
import logging
from typing import Optional
2022-07-27 17:50:18 +02:00
from requests import Response, Session as BaseSession
from requests.adapters import HTTPAdapter
from requests.exceptions import RequestException
from urllib3.util import Retry
2023-02-26 01:27:00 +01:00
logger = logging.getLogger(__name__)
DEFAULT_TIMEOUT = 5
class TimeoutHTTPAdapter(HTTPAdapter):
timeout: int = DEFAULT_TIMEOUT
def __init__(self, *args, **kwargs):
if "timeout" in kwargs:
self.timeout = kwargs["timeout"]
del kwargs["timeout"]
super().__init__(*args, **kwargs)
def send(self, request, *args, **kwargs):
if "timeout" not in kwargs:
kwargs["timeout"] = self.timeout
return super().send(request, *args, **kwargs)
def default_retry(max_retries: int = 5):
return Retry(
total=max_retries,
backoff_factor=2,
status_forcelist=[413, 429, 500, 502, 503, 504],
)
class Session(BaseSession):
base_url: Optional[str]
def __init__(
self,
base_url: Optional[str] = None,
retry: Optional[Retry] = None,
):
super().__init__()
self.base_url = base_url
adapter = TimeoutHTTPAdapter(max_retries=retry)
self.mount("http://", adapter)
self.mount("https://", adapter)
def request(self, method, url, *args, **kwargs):
"""Send the request after generating the complete URL."""
url = self.create_url(url)
return super().request(method, url, *args, **kwargs)
def create_url(self, url):
"""Create the URL based off this partial path."""
if self.base_url is None:
return url
return f"{self.base_url.rstrip('/')}/{url.lstrip('/')}"
# pylint: disable=too-few-public-methods
class AbstractApiClient:
session: Session
base_url: str
def __init__(
self,
base_url: str,
retry: Optional[Retry] = None,
):
self.base_url = base_url
self.session = Session(
base_url=base_url,
retry=retry,
)
def _request(
self,
method,
url,
**kwargs,
) -> Response:
try:
response = self.session.request(method, url, **kwargs)
response.raise_for_status()
return response
except RequestException as exception:
logger.error(exception)
raise exception