41 lines
1.2 KiB
Python
41 lines
1.2 KiB
Python
import logging
|
|
|
|
import aiohttp
|
|
|
|
logger = logging.getLogger("source")
|
|
|
|
|
|
class ApiSource:
|
|
DEFAULT_USER_AGENT = (
|
|
"Mozilla/5.0 (X11; Linux x86_64) "
|
|
"AppleWebKit/537.36 (KHTML, like Gecko) "
|
|
"Chrome/126.0.0.0 Safari/537.36"
|
|
)
|
|
DEFAULT_TIMEOUT = 30.0
|
|
|
|
def __init__(
|
|
self,
|
|
base_url,
|
|
user_agent: str = DEFAULT_USER_AGENT,
|
|
timeout: float = DEFAULT_TIMEOUT,
|
|
cookies: dict[str, str] | None = None,
|
|
headers: dict[str, str] | None = None,
|
|
):
|
|
self._base_url = base_url
|
|
self._user_agent = user_agent
|
|
self._timeout = timeout
|
|
self._cookies = cookies
|
|
self._headers = headers
|
|
|
|
async def request(self, endpoint: str) -> str:
|
|
url = f"{self._base_url}/{endpoint}"
|
|
logger.info(url)
|
|
headers = {"User-Agent": self._user_agent, **(self._headers or {})}
|
|
async with aiohttp.ClientSession(
|
|
headers=headers,
|
|
cookies=self._cookies,
|
|
raise_for_status=True,
|
|
) as session:
|
|
async with session.request("GET", url, timeout=self._timeout) as response:
|
|
return await response.text()
|