AIOHTTP

Extensions and helpers for building AIOHTTP-based applications.

Middlewares

async kw.platform.aiohttp.middlewares.user_agent_middleware(request, handler)[source]

Validate client’s User-Agent header and modify response based on that.

If the User-Agent header is invalid, there are three possible outcomes:

  1. The current time is less then settings.KIWI_REQUESTS_SLOWDOWN_DATETIME, do nothing in this case.

  2. The current time is less then settings.KIWI_REQUESTS_RESTRICT_DATETIME, slow down the response twice the normal responce time.

  3. The current time is more then settings.KIWI_REQUESTS_RESTRICT_DATETIME, refuse the request, return HTTP 400 to the client.

Usage:

from aiohttp import web

from kw.platform.aiohttp.middlewares import user_agent_middleware

app = web.Application(middlewares=[user_agent_middleware])

Warning

The middleware slows down requests by calling asyncio.sleep() (in the time frame when requests with invalid user-agent are being delayed). This can increase busyness and overload a service.

Monkey Patching

kw.platform.aiohttp.monkey.patch()[source]

Apply all patches for aiohttp module.

This will automatically apply:

  • kw.platform.aiohttp.patch.patch_with_user_agent()

kw.platform.aiohttp.monkey.patch_with_user_agent(user_agent=None)[source]

Patch aiohttp.ClientSession._request() with User-Agent.

In case User-Agent header has not been provided directly to request. Add User-Agent string constructed by kw.platform.aiohttp.patch.construct_user_agent() as User-Agent header.

Parameters

user_agent – (optional) User-Agent string that will be used as User-Agent header.

Session

class kw.platform.aiohttp.session.KiwiClientSession(*, connector: Optional[aiohttp.connector.BaseConnector] = None, loop: Optional[asyncio.events.AbstractEventLoop] = None, cookies: Union[Iterable[Tuple[str, BaseCookie[str]]], Mapping[str, BaseCookie[str]], BaseCookie[str], None] = None, headers: Union[Mapping[str, str], multidict._multidict.CIMultiDict, multidict._multidict.CIMultiDictProxy, None] = None, skip_auto_headers: Optional[Iterable[str]] = None, auth: Optional[aiohttp.helpers.BasicAuth] = None, json_serialize: Callable[[Any], str] = <function dumps>, request_class: Type[aiohttp.client_reqrep.ClientRequest] = <class 'aiohttp.client_reqrep.ClientRequest'>, response_class: Type[aiohttp.client_reqrep.ClientResponse] = <class 'aiohttp.client_reqrep.ClientResponse'>, ws_response_class: Type[aiohttp.client_ws.ClientWebSocketResponse] = <class 'aiohttp.client_ws.ClientWebSocketResponse'>, version: aiohttp.http_writer.HttpVersion = HttpVersion(major=1, minor=1), cookie_jar: Optional[aiohttp.abc.AbstractCookieJar] = None, connector_owner: bool = True, raise_for_status: bool = False, read_timeout: Union[float, object] = <object object>, conn_timeout: Optional[float] = None, timeout: Union[object, aiohttp.client.ClientTimeout] = <object object>, auto_decompress: bool = True, trust_env: bool = False, requote_redirect_url: bool = True, trace_configs: Optional[List[aiohttp.tracing.TraceConfig]] = None)[source]

Custom aiohttp.ClientSession with all patches applied.

Usage:

from kw.platform.aiohttp import KiwiClientSession
async with KiwiClientSession() as c:
    await c.get('https://kiwi.com')