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:
The current time is less then
settings.KIWI_REQUESTS_SLOWDOWN_DATETIME
, do nothing in this case.The current time is less then
settings.KIWI_REQUESTS_RESTRICT_DATETIME
, slow down the response twice the normal responce time.The current time is more then
settings.KIWI_REQUESTS_RESTRICT_DATETIME
, refuse the request, returnHTTP 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.patch.patch_with_sentry()
-
kw.platform.aiohttp.monkey.
patch_with_sentry
(sunset_header=True, deprecated_usage_header=True)[source]¶ Patch
aiohttp.ClientSession._request()
to create events in Sentry.If the HTTP response contains, for example, the
Sunset
HTTP header, an event is sent to Sentry containing details about the sunset.- Parameters
sunset_header – (optional) Whether to report the presence of the
Sunset
header,True
by default.deprecated_usage_header – (optional) Whether to report the presence of the
Deprecated-Usage
header,True
by default.
-
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[Union[str, multidict._multidict.istr], 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 client: await client.get('https://kiwi.com')
Helpers and Utils¶
-
kw.platform.aiohttp.utils.
mandatory_user_agent
(handler)[source]¶ A decorator for the validation that the request to the decorated handler is compliant with KW-RFC-22. See
kw.platform.aiohttp.middlewares.user_agent_middleware()
for more information about what happens to the request if the User-Agent header is not compliant.Usage:
@mandatory_user_agent async def handle(request): return web.json_response(message="Hello, Wordl!")
-
kw.platform.aiohttp.utils.
set_sunset
(response, when=None, info_url=None)[source]¶ Update aiohttp response object with the
Sunset
HTTP header.- Parameters
response (
aiohttp.web.Response
) – Response object to update.when (datetime) – When the Sunset date arrives, must be UTC.
info_url (str) – URL to a page with more information about the Sunset.
- Return type
aiohttp.web.Response
-
kw.platform.aiohttp.utils.
sunset
(when=None, info_url=None)[source]¶ A decorator for deprecating views by setting the
Sunset
HTTP header.Read about the
Sunset
header in rfc8594.Usage:
@sunset(when=datetime(2019, 8, 1, 10, 0, 0)) async def index(request): return web.Response("Hello, world!")
- Parameters
when (datetime) – When the Sunset date arrives, must be UTC.
info_url (str) – URL to a page with more information about the Sunset.