91 lines
2.5 KiB
Python
91 lines
2.5 KiB
Python
"""# Common HTTP/REST clients interface
|
|
"""
|
|
|
|
import abc
|
|
import logging
|
|
from collections.abc import Sequence
|
|
from typing import Any
|
|
|
|
import bs4
|
|
import lxml.html
|
|
import requests
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
API_ERROR_KEY = 'error'
|
|
|
|
|
|
class ApiError(RuntimeError):
|
|
pass
|
|
|
|
|
|
class AbstractClient(abc.ABC):
|
|
SECRETS: Sequence[str] = ()
|
|
|
|
def __init__(self, session: requests.Session):
|
|
assert isinstance(session, requests.Session)
|
|
self.session = session
|
|
|
|
def fetch_or_none(
|
|
self,
|
|
url: str,
|
|
params=None,
|
|
**kwargs,
|
|
) -> requests.Response | None:
|
|
r = self._fetch(url, params, **kwargs)
|
|
if r.status_code == 404:
|
|
return None
|
|
return r
|
|
|
|
def fetch(self, url: str, params=None, **kwargs) -> requests.Response:
|
|
r = self._fetch(url, params, **kwargs)
|
|
r.raise_for_status()
|
|
return r
|
|
|
|
def _fetch(self, url: str, params=None, **kwargs) -> requests.Response:
|
|
method = 'GET'
|
|
if 'method' in kwargs:
|
|
method = kwargs['method']
|
|
del kwargs['method']
|
|
kwargs.setdefault('headers', {}).setdefault('Origin', url) # TODO?
|
|
return self.session.request(
|
|
method,
|
|
url,
|
|
params=params,
|
|
allow_redirects=True,
|
|
**kwargs,
|
|
)
|
|
|
|
def fetch_text(self, url: str, params=None, **kwargs) -> str:
|
|
return self.fetch(url, params, **kwargs).text
|
|
|
|
def fetch_lxml_soup(
|
|
self,
|
|
url: str,
|
|
params=None,
|
|
**kwargs,
|
|
) -> None | bs4.BeautifulSoup:
|
|
kwargs.setdefault('headers', {}).setdefault('Accept', 'text/html')
|
|
text = self.fetch_text(url, params, **kwargs)
|
|
if text is None:
|
|
return None
|
|
return lxml.html.document_fromstring(text)
|
|
|
|
def fetch_soup(self, url: str, params=None, **kwargs) -> None | bs4.BeautifulSoup:
|
|
kwargs.setdefault('headers', {}).setdefault('Accept', 'text/html')
|
|
text = self.fetch_text(url, params, **kwargs)
|
|
if text is None:
|
|
return None
|
|
return bs4.BeautifulSoup(text, 'html.parser')
|
|
|
|
def fetch_json(self, url: str, params=None, **kwargs) -> None | dict[str, Any]:
|
|
kwargs.setdefault('headers', {}).setdefault('Accept', 'application/json')
|
|
|
|
response = self.fetch(url, params, **kwargs)
|
|
loaded_json = response.json()
|
|
if API_ERROR_KEY in loaded_json:
|
|
msg = f'Error from endpoint: {loaded_json[API_ERROR_KEY]}'
|
|
raise ApiError(msg)
|
|
return loaded_json
|