from __future__ import annotations import json from typing import Any, Mapping, MutableMapping from requests.structures import CaseInsensitiveDict from .cookies import CookieJar, cookiejar_from_dict from .errors import BadRequest from .types import JSONLike __all__ = ["Request"] class Request: _content_type: str = "application/json" def __init__(self, event: Mapping[str, Any]): self.multi_value_query_parameters: Mapping[str, list[Any]] = event.get("multiValueQueryStringParameters", {}) self.query_parameters: Mapping[str, Any] = event.get("queryStringParameters", {}) self.headers: MutableMapping[str, Any] = CaseInsensitiveDict(event.get("headers", {})) self.cookies: CookieJar = self._get_cookies() self.path: str = event["path"] self.method: str = event["httpMethod"] self.body: str | None = event.get("body") def _get_cookies(self) -> CookieJar: cookies_dict = {} cookies_str = self.headers.get("cookie", "") for cookie_str in cookies_str.split(";"): cookie_str = cookie_str.strip() if not cookie_str: continue key, value = cookie_str.split("=", 1) cookies_dict[key] = value return cookiejar_from_dict(cookies_dict, CookieJar()) def _parse_body(self) -> JSONLike | None: content_type = self.headers.get("Content-Type", "") if content_type.lower() != self.__class__._content_type: raise BadRequest(f"Unexpected content type '{content_type}'") if self.body is None: raise BadRequest("Can't parse empty body") try: return json.loads(self.body) except Exception as e: raise BadRequest("Can't deserialize request body") from e @property def json(self) -> JSONLike | None: if not hasattr(self, "_json_body"): setattr(self, "_json_body", self._parse_body()) return getattr(self, "_json_body") @property def host(self) -> str: return self.headers["Host"] @property def schema(self) -> str: return self.headers.get("X-Forwarded-Proto", "http") def __repr__(self): attrs = ("method", "path", "query_parameters", "body", "headers") return f"<{self.__class__.__qualname__} {', '.join(f'{attr}: {getattr(self, attr)}' for attr in attrs)}>"