mbeliaev/recorder

CHANGES

@@ -1,6 +1,7 @@
 0.22.0
 ------
 
+* [BETA] Added possibility to record responses to YAML files via `@_recorder.record(file_path="out.yml")` decorator.
 * Add `passthrough` argument to `BaseResponse` object. See #557
 
 0.21.0

responses/_recorder.py

@@ -0,0 +1,75 @@
+from functools import wraps
+from typing import TYPE_CHECKING
+
+if TYPE_CHECKING:
+    from typing import Any
+    from typing import Callable
+    from typing import Type
+    from typing import Union
+    import os
+    from responses import FirstMatchRegistry
+    from responses import HTTPAdapter
+    from responses import PreparedRequest
+    from responses import models
+    from responses import _F
+
+from responses import RequestsMock
+from responses import Response
+from responses import _real_send
+from responses.registries import OrderedRegistry
+
+
+class Recorder(RequestsMock):
+    def __init__(
+        self,
+        target: str = "requests.adapters.HTTPAdapter.send",
+        registry: "Type[FirstMatchRegistry]" = OrderedRegistry,
+    ) -> None:
+        super().__init__(target=target, registry=registry)
+
+    def reset(self) -> None:
+        self._registry = OrderedRegistry()
+
+    def record(
+        self, *, file_path: "Union[str, bytes, os.PathLike[Any]]" = "response.toml"
+    ) -> "Union[Callable[[_F], _F], _F]":
+        def deco_record(function: "_F") -> "Callable[..., Any]":
+            @wraps(function)
+            def wrapper(*args: "Any", **kwargs: "Any") -> "Any":  # type: ignore[misc]
+                with self:
+                    ret = function(*args, **kwargs)
+                    with open(file_path, "w") as file:
+                        self.get_registry()._dump(file)
+
+                    return ret
+
+            return wrapper
+
+        return deco_record
+
+    def _on_request(
+        self,
+        adapter: "HTTPAdapter",
+        request: "PreparedRequest",
+        **kwargs: "Any",
+    ) -> "models.Response":
+        # add attributes params and req_kwargs to 'request' object for further match comparison
+        # original request object does not have these attributes
+        request.params = self._parse_request_params(request.path_url)  # type: ignore[attr-defined]
+        request.req_kwargs = kwargs  # type: ignore[attr-defined]
+        requests_response = _real_send(adapter, request, **kwargs)
+        responses_response = Response(
+            method=str(request.method),
+            url=str(requests_response.request.url),
+            status=requests_response.status_code,
+            body=requests_response.text,
+        )
+        self._registry.add(responses_response)
+        return requests_response
+
+    def stop(self, allow_assert: bool = True) -> None:
+        super().stop(allow_assert=False)
+
+
+recorder = Recorder()
+record = recorder.record

responses/registries.py

@@ -1,11 +1,17 @@
 import copy
 from typing import TYPE_CHECKING
+from typing import Any
+from typing import Dict
 from typing import List
 from typing import Optional
 from typing import Tuple
 
+import toml as _toml
+
 if TYPE_CHECKING:  # pragma: no cover
     # import only for linter run
+    import io
+
     from requests import PreparedRequest
 
     from responses import BaseResponse
@@ -73,6 +79,31 @@ def replace(self, response: "BaseResponse") -> "BaseResponse":
         self.registered[index] = response
         return response
 
+    def _dump(self, destination: "io.IOBase") -> None:
+        data: Dict[str, Any] = {"responses": []}
+        for rsp in self.registered:
+            try:
+                content_length = rsp.auto_calculate_content_length  # type: ignore[attr-defined]
+                data["responses"].append(
+                    {
+                        "response": {
+                            "method": rsp.method,
+                            "url": rsp.url,
+                            "body": rsp.body,  # type: ignore[attr-defined]
+                            "status": rsp.status,  # type: ignore[attr-defined]
+                            "headers": rsp.headers,
+                            "content_type": rsp.content_type,
+                            "auto_calculate_content_length": content_length,
+                        }
+                    }
+                )
+            except AttributeError as exc:
+                raise AttributeError(
+                    "Cannot dump response object."
+                    "Probably you use custom Response object that misses required aatributes"
+                ) from exc
+        _toml.dump(data, destination)
+
 
 class OrderedRegistry(FirstMatchRegistry):
     def find(

setup.py

@@ -19,6 +19,8 @@
 install_requires = [
     "requests>=2.0,<3.0",
     "urllib3>=1.25.10",
+    "toml",
+    "types-toml",
     "typing_extensions; python_version < '3.8'",
 ]