Coverage for src/zapy/requests/requester.py: 94%
99 statements
« prev ^ index » next coverage.py v7.4.1, created at 2024-02-10 19:35 +0000
« prev ^ index » next coverage.py v7.4.1, created at 2024-02-10 19:35 +0000
1import asyncio
2import inspect
3from dataclasses import dataclass, field
4from typing import Any, Callable, List
6import httpx
8from zapy.base import Metadata
9from zapy.store import Store, use_store
10from zapy.templating.traceback import annotate_traceback
11from zapy.test import TestResult, run_tests
12from zapy.utils import functools
14from .context import ZapyRequestContext
15from .converter import RequestConverter
16from .exceptions import RenderLocationError, error_location
17from .hooks import RequestHook, use_global_hook
18from .models import HttpxArguments, HttpxResponse, ZapyRequest
21@dataclass
22class RequesterResponse:
23 response: httpx.Response
24 test_result: List[TestResult] = field(default_factory=list)
27_http_request_signature = inspect.signature(httpx.Client.build_request).parameters
30class Requester:
32 def __init__(self, zapy_request: ZapyRequest, converter: RequestConverter, client: httpx.AsyncClient):
33 self.zapy_request = zapy_request
34 self.converter = converter
35 self.client = client
36 self.__hook_context = {
37 Metadata: self.zapy_request.metadata,
38 ZapyRequest: self.zapy_request,
39 }
41 async def make_request(self) -> RequesterResponse:
42 httpx_args = self.converter.build_httpx_args()
44 # Hook: pre_request
45 await self._invoke_hooks_pre_request(httpx_args)
47 # send request
48 request_parameters, rest_parameters = self._split_parameters(httpx_args)
49 request = self.client.build_request(**request_parameters)
50 response = await self.client.send(request, **rest_parameters)
52 # Hook: post_request
53 try:
54 await self._invoke_hooks_post_request(response)
55 except RenderLocationError as ex:
56 ex.context["response"] = response
57 raise
59 response_wrapper = RequesterResponse(response)
61 # Hook: test
62 if self.request_hooks.test:
63 response_wrapper.test_result = self._run_test(
64 httpx_args=httpx_args,
65 request=request,
66 response=response,
67 )
69 return response_wrapper
71 @error_location("pre_request")
72 async def _invoke_hooks_pre_request(self, httpx_args: HttpxArguments) -> None:
73 try:
74 await self.__call_hook(use_global_hook().pre_request, httpx_args)
75 await self.__call_hook(self.request_hooks.pre_request, httpx_args)
76 except BaseException as e:
77 annotate_traceback(e, self.converter.script, location="hook")
78 raise e
80 @error_location("post_request")
81 async def _invoke_hooks_post_request(self, response: httpx.Response) -> None:
82 try:
83 await self.__call_hook(use_global_hook().post_request, response)
84 await self.__call_hook(self.request_hooks.post_request, response)
85 except Exception as e:
86 annotate_traceback(e, self.converter.script, location="hook")
87 raise e
89 def _run_test(
90 self, httpx_args: HttpxArguments, request: httpx.Request, response: HttpxResponse
91 ) -> list[TestResult]:
92 if self.request_hooks.test is None: 92 ↛ 93line 92 didn't jump to line 93, because the condition on line 92 was never true
93 return []
95 class RequestMeta(type):
96 def __new__(cls, name: str, bases: tuple, attrs: dict) -> type:
97 attrs["httpx_args"] = httpx_args
98 attrs["request"] = request
99 attrs["response"] = response
100 return super().__new__(cls, name, bases, attrs)
102 class DecoratedClass(self.request_hooks.test, metaclass=RequestMeta): # type: ignore[name-defined]
103 pass
105 test_result = run_tests(DecoratedClass).as_list()
106 return test_result
108 def _split_parameters(self, httpx_args: HttpxArguments) -> tuple[dict, dict]:
109 request_parameters, rest_parameters = {}, {}
110 for k, v in httpx_args.items():
111 if k in _http_request_signature: 111 ↛ 114line 111 didn't jump to line 114, because the condition on line 111 was never false
112 request_parameters[k] = v
113 else:
114 rest_parameters[k] = v
115 return request_parameters, rest_parameters
117 async def __call_hook(self, hook: Callable, *args: Any) -> Any:
118 result = functools.call_with_signature(hook, *args, kwargs=self.__hook_context)
119 if asyncio.iscoroutine(result): 119 ↛ 121line 119 didn't jump to line 121, because the condition on line 119 was never false
120 return await result
121 return result
123 @property
124 def request_hooks(self) -> RequestHook:
125 return self.converter.request_hooks
128async def send_request(
129 zapy_request: ZapyRequest,
130 *,
131 store: Store | None = None,
132 logger: Callable = print,
133 client: httpx.AsyncClient | None = None,
134) -> RequesterResponse:
135 if store is None:
136 store = use_store()
137 ctx = ZapyRequestContext(
138 store=store,
139 logger=logger,
140 )
141 _client = client or httpx.AsyncClient(follow_redirects=True, timeout=None)
142 try:
143 request = build_request(zapy_request, ctx, client=_client)
144 return await request.make_request()
145 finally:
146 if client is None: 146 ↛ exitline 146 didn't return from function 'send_request', because the return on line 144 wasn't executed
147 await _client.aclose()
150def build_request(zapy_request: ZapyRequest, ctx: ZapyRequestContext, client: httpx.AsyncClient) -> Requester:
151 converter = RequestConverter(zapy_request, ctx)
152 requester = Requester(zapy_request, converter, client)
154 return requester