Response now has async read() method instead of data property + reorganize Response's typings

Author: igorcoding

extapi/http/abc.py

@@ -5,10 +5,12 @@
 
 from .types import RequestData, Response, StrOrURL
 
+T_co = TypeVar("T_co", covariant=True)
+T_contr = TypeVar("T_contr", contravariant=True)
 T = TypeVar("T")
 
 
-class AbstractExecutor(Generic[T], metaclass=abc.ABCMeta):
+class AbstractExecutor(Generic[T_co], metaclass=abc.ABCMeta):
     async def start(self) -> None:
         return None
 
@@ -22,14 +24,14 @@ async def __aenter__(self) -> Self:
     async def __aexit__(self, exc_type, exc_val, exc_tb) -> None:
         await self.close()
 
-    def generalize(self) -> "AbstractExecutor[T]":
+    def generalize(self) -> "AbstractExecutor[T_co]":
         return self
 
     @abc.abstractmethod
     async def execute(
         self,
         request: RequestData,
-    ) -> Response[T]:
+    ) -> Response[T_co]:
         raise NotImplementedError  # pragma: no cover
 
     async def get(
@@ -42,7 +44,7 @@ async def get(
         headers: CIMultiDict | None = None,
         timeout: Any | float | None = None,
         **kwargs,
-    ) -> Response[T]:
+    ) -> Response[T_co]:
         return await self.execute(
             RequestData(
                 method="GET",
@@ -66,7 +68,7 @@ async def post(
         headers: CIMultiDict | None = None,
         timeout: Any | float | None = None,
         **kwargs,
-    ) -> Response[T]:
+    ) -> Response[T_co]:
         return await self.execute(
             RequestData(
                 method="POST",
@@ -90,7 +92,7 @@ async def delete(
         headers: CIMultiDict | None = None,
         timeout: Any | float | None = None,
         **kwargs,
-    ) -> Response[T]:
+    ) -> Response[T_co]:
         return await self.execute(
             RequestData(
                 method="DELETE",
@@ -114,7 +116,7 @@ async def put(
         headers: CIMultiDict | None = None,
         timeout: Any | float | None = None,
         **kwargs,
-    ) -> Response[T]:
+    ) -> Response[T_co]:
         return await self.execute(
             RequestData(
                 method="PUT",
@@ -138,7 +140,7 @@ async def patch(
         headers: CIMultiDict | None = None,
         timeout: Any | float | None = None,
         **kwargs,
-    ) -> Response[T]:
+    ) -> Response[T_co]:
         return await self.execute(
             RequestData(
                 method="PATCH",
@@ -154,8 +156,10 @@ async def patch(
 
 
 @runtime_checkable
-class Retryable(Protocol[T]):
-    async def need_retry(self, response: Response[T]) -> tuple[bool, float | None]: ...
+class Retryable(Protocol[T_contr]):
+    async def need_retry(
+        self, response: Response[T_contr]
+    ) -> tuple[bool, float | None]: ...
 
 
 @runtime_checkable

extapi/http/addons/log.py

@@ -2,41 +2,57 @@
 import logging
 from typing import Generic, TypeVar
 
+from yarl import URL
+
 from extapi.http.abc import Addon
 from extapi.http.types import HttpExecuteError, RequestData, Response
 
 T = TypeVar("T")
 
 
 class LoggingAddon(Addon[T], Generic[T]):
-    def __init__(self):
+    def __init__(self, *, log_params: bool = True):
         self._logger = logging.getLogger("extapi.http.addons.log")
+        self._log_params = log_params
+
+    def _get_url(self, request: RequestData) -> URL:
+        url_ = URL(request.url) if not isinstance(request.url, URL) else request.url
+
+        if self._log_params and request.params:
+            return url_.with_query(request.params)
+
+        return url_
 
     async def before_request(self, request: RequestData) -> None:
-        self._logger.debug("executing request %s %s", request.method, str(request.url))
+        url = self._get_url(request)
+        self._logger.debug("executing request %s %s", request.method, str(url))
 
     async def process_response(
         self, request: RequestData, response: Response[T]
     ) -> Response[T]:
+        url = self._get_url(request)
+
         logger_method = (
             self._logger.debug if response.status < 500 else self._logger.error
         )
 
         logger_method(
             "received response %s %s -> status=%s",
             request.method,
-            str(request.url),
+            str(url),
             response.status if response is not None else "unknown",
         )
 
         return response
 
     async def process_error(self, request: RequestData, error: Exception) -> None:
+        url = self._get_url(request)
+
         if isinstance(error, TimeoutError):
             self._logger.error(
                 "timeout error for request %s %s failed with error %s(%s)",
                 request.method,
-                str(request.url),
+                str(url),
                 type(error).__name__,
                 str(error),
             )
@@ -46,7 +62,7 @@ async def process_error(self, request: RequestData, error: Exception) -> None:
             self._logger.error(
                 "request %s %s failed with error %s(%s)",
                 request.method,
-                str(request.url),
+                str(url),
                 type(error).__name__,
                 error,
             )
@@ -56,12 +72,17 @@ class VerboseLoggingExecutor(LoggingAddon[T], Generic[T]):
     def __init__(
         self,
         *,
+        log_params: bool = True,
+        log_response_data: bool = True,
         truncate_response_data: int | None = 1024,
     ):
-        super().__init__()
+        super().__init__(log_params=log_params)
+        self._log_response_data = log_response_data
         self._truncate_response_data = truncate_response_data
 
     async def before_request(self, request: RequestData) -> None:
+        url = self._get_url(request)
+
         json = request.json
         if json is not None:
             if isinstance(json, bytes):
@@ -71,7 +92,7 @@ async def before_request(self, request: RequestData) -> None:
         self._logger.debug(
             "executing request %s %s with params=%s json=%s data=%s timeout=%s",
             request.method,
-            str(request.url),
+            str(url),
             request.params,
             json,
             request.data,
@@ -81,18 +102,22 @@ async def before_request(self, request: RequestData) -> None:
     async def process_response(
         self, request: RequestData, response: Response[T]
     ) -> Response[T]:
+        url = self._get_url(request)
+
         logger_method = (
             self._logger.debug if response.status < 500 else self._logger.error
         )
-
-        resp_body = response.data.decode("utf-8") if response.has_data else None
-        if resp_body is not None and self._truncate_response_data is not None:
-            resp_body = resp_body[: self._truncate_response_data]
+        resp_body: str | None = None
+        if self._log_response_data:
+            resp_body_bytes = await response.read()
+            if self._truncate_response_data is not None:
+                resp_body_bytes = resp_body_bytes[: self._truncate_response_data]
+            resp_body = resp_body_bytes.decode("utf-8")
 
         logger_method(
             "received response %s %s -> status=%s headers=%s body=%s",
             request.method,
-            str(request.url),
+            str(url),
             response.status,
             response.headers,
             resp_body,

extapi/http/addons/retry.py

@@ -7,29 +7,35 @@
 
 
 class Retry5xxAddon(Retryable[T], Generic[T]):
-    __slots__ = ("_retry_timeout",)
+    __slots__ = ("_default_timeout",)
 
-    def __init__(self, *, retry_timeout: float | None = None):
-        self._retry_timeout = retry_timeout
+    def __init__(self, *, default_timeout: float | None = None):
+        self._default_timeout = default_timeout
 
     async def need_retry(self, response: Response[T]) -> tuple[bool, float | None]:
         if response.status >= 500:
-            return True, self._retry_timeout
+            return True, self._default_timeout
 
         return False, None
 
 
 class Retry429Addon(Retryable[T], Generic[T]):
+    __slots__ = ("_default_timeout",)
+
+    def __init__(self, *, default_timeout: float | None = None):
+        self._default_timeout = default_timeout
+
     async def need_retry(self, response: Response[T]) -> tuple[bool, float | None]:
         if response.status != 429:
             return False, None
 
         retry_after_s = response.headers.get("retry-after")
         if retry_after_s is not None:
+            retry_after: float | None
             try:
                 retry_after = float(retry_after_s)
             except ValueError:
-                retry_after = None
+                retry_after = self._default_timeout
 
             return True, retry_after
 

extapi/http/backends/aiohttp.py

@@ -3,21 +3,27 @@
 import aiohttp
 
 from extapi.http.abc import AbstractExecutor
-from extapi.http.types import Closable, RequestData, Response
+from extapi.http.types import BackendResponseProtocol, RequestData, Response
 
 
-class AiohttpResponseWrap(Closable):
-    __slots__ = ("original",)
+class AiohttpResponseWrap(BackendResponseProtocol[aiohttp.ClientResponse]):
+    __slots__ = ("_original",)
 
     def __init__(self, response: aiohttp.ClientResponse):
-        self.original = response
+        self._original = response
+
+    def original(self) -> aiohttp.ClientResponse:
+        return self._original
 
     async def close(self) -> None:
-        self.original.release()
-        await self.original.wait_for_close()
+        self._original.release()
+        await self._original.wait_for_close()
+
+    async def read(self) -> bytes:
+        return await self._original.read()
 
 
-class AiohttpStreamingExecutor(AbstractExecutor[AiohttpResponseWrap]):
+class AiohttpExecutor(AbstractExecutor[aiohttp.ClientResponse]):
     __slots__ = (
         "_ssl",
         "_session",
@@ -35,7 +41,7 @@ def __init__(
     async def close(self):
         await self._session.close()
 
-    async def execute(self, request: RequestData) -> Response[AiohttpResponseWrap]:
+    async def execute(self, request: RequestData) -> Response[aiohttp.ClientResponse]:
         timeout = request.timeout or self._default_timeout
 
         response = await self._session.request(
@@ -50,17 +56,9 @@ async def execute(self, request: RequestData) -> Response[AiohttpResponseWrap]:
             **request.kwargs,
         )
 
-        return Response[AiohttpResponseWrap](
+        return Response[aiohttp.ClientResponse](
             url=request.url,
             status=response.status,
             headers=response.headers.copy(),
             backend_response=AiohttpResponseWrap(response),
         )
-
-
-class AiohttpExecutor(AiohttpStreamingExecutor):
-    async def execute(self, request: RequestData) -> Response[AiohttpResponseWrap]:
-        resp = await super().execute(request)
-        async with resp:
-            resp.set_data(await resp.backend_response.original.read())
-        return resp