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() 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)