75 lines
2.1 KiB
Python
75 lines
2.1 KiB
Python
import datetime
|
|
import logging
|
|
from typing import Any, Literal
|
|
|
|
import aiohttp
|
|
from playwright.async_api import async_playwright
|
|
|
|
logger = logging.getLogger("source")
|
|
|
|
RequestMode = Literal["base", "browser"]
|
|
|
|
DEFAULT_REQUEST_MODE: RequestMode = "base"
|
|
DEFAULT_METHOD: str = "GET"
|
|
|
|
|
|
class ApiSource:
|
|
DEFAULT_TIMEOUT = 30.0
|
|
|
|
def __init__(
|
|
self,
|
|
base_url,
|
|
timeout: float = DEFAULT_TIMEOUT,
|
|
cookies: dict[str, str] | None = None,
|
|
headers: dict[str, str] | None = None,
|
|
):
|
|
self._base_url = base_url
|
|
self._timeout = timeout
|
|
self._cookies = cookies
|
|
self._headers = headers
|
|
|
|
async def _base_request(
|
|
self,
|
|
url: str,
|
|
method: str = DEFAULT_METHOD,
|
|
data: Any | None = None,
|
|
) -> str:
|
|
async with aiohttp.ClientSession(
|
|
headers=self._headers,
|
|
cookies=self._cookies,
|
|
raise_for_status=True,
|
|
) as session:
|
|
async with session.request(method, url, timeout=self._timeout, data=data) as response:
|
|
return await response.text()
|
|
|
|
# pylint:disable=unused-argument
|
|
async def _browser_request(
|
|
self,
|
|
url: str,
|
|
method: str = DEFAULT_METHOD,
|
|
data: Any | None = None,
|
|
) -> str:
|
|
async with async_playwright() as p:
|
|
browser = await p.chromium.launch()
|
|
page = await browser.new_page()
|
|
await page.goto(url, timeout=datetime.timedelta(seconds=self._timeout), wait_until="networkidle")
|
|
content = await page.content()
|
|
await browser.close()
|
|
return content
|
|
|
|
async def request(
|
|
self,
|
|
endpoint: str,
|
|
/,
|
|
mode: RequestMode = DEFAULT_REQUEST_MODE,
|
|
method: str = DEFAULT_METHOD,
|
|
data: Any | None = None,
|
|
) -> str:
|
|
if endpoint.startswith("https:"):
|
|
url = endpoint
|
|
else:
|
|
url = f"{self._base_url}/{endpoint}"
|
|
logger.info(url)
|
|
fun = getattr(self, f"_{mode}_request")
|
|
return await fun(url, method, data)
|