38 lines
1.3 KiB
Python
38 lines
1.3 KiB
Python
import logging
|
|
from collections.abc import AsyncIterator
|
|
from contextlib import asynccontextmanager
|
|
from typing import Optional, List, Tuple, Any, Dict
|
|
|
|
import aiohttp
|
|
from yarl import URL
|
|
|
|
from pyhon import const
|
|
from pyhon.connection.handler.base import ConnectionHandler
|
|
from pyhon.typedefs import Callback
|
|
|
|
_LOGGER = logging.getLogger(__name__)
|
|
|
|
|
|
class HonAuthConnectionHandler(ConnectionHandler):
|
|
_HEADERS = {"user-agent": const.USER_AGENT}
|
|
|
|
def __init__(self, session: Optional[aiohttp.ClientSession] = None) -> None:
|
|
super().__init__(session)
|
|
self._called_urls: List[Tuple[int, str]] = []
|
|
|
|
@property
|
|
def called_urls(self) -> List[Tuple[int, str]]:
|
|
return self._called_urls
|
|
|
|
@called_urls.setter
|
|
def called_urls(self, called_urls: List[Tuple[int, str]]) -> None:
|
|
self._called_urls = called_urls
|
|
|
|
@asynccontextmanager
|
|
async def _intercept(
|
|
self, method: Callback, url: str | URL, *args: Any, **kwargs: Dict[str, Any]
|
|
) -> AsyncIterator[aiohttp.ClientResponse]:
|
|
kwargs["headers"] = kwargs.pop("headers", {}) | self._HEADERS
|
|
async with method(url, *args, **kwargs) as response:
|
|
self._called_urls.append((response.status, str(response.request_info.url)))
|
|
yield response
|