1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
|
from typing import Any
from typing import Optional
from jsonschema_path import SchemaPath
from openapi_core.protocols import Request
from openapi_core.protocols import Response
from openapi_core.unmarshalling.response.datatypes import (
ResponseUnmarshalResult,
)
from openapi_core.unmarshalling.response.protocols import ResponseUnmarshaller
from openapi_core.unmarshalling.response.types import ResponseUnmarshallerType
class ResponseUnmarshallingProcessor:
def __init__(
self,
spec: SchemaPath,
response_unmarshaller_cls: ResponseUnmarshallerType,
**unmarshaller_kwargs: Any
) -> None:
self.spec = spec
self.response_unmarshaller_cls = response_unmarshaller_cls
self.unmarshaller_kwargs = unmarshaller_kwargs
self._response_unmarshaller_cached: Optional[ResponseUnmarshaller] = (
None
)
@property
def response_unmarshaller(self) -> ResponseUnmarshaller:
if self._response_unmarshaller_cached is None:
self._response_unmarshaller_cached = (
self.response_unmarshaller_cls(
self.spec, **self.unmarshaller_kwargs
)
)
return self._response_unmarshaller_cached
def process(
self, request: Request, response: Response
) -> ResponseUnmarshalResult:
return self.response_unmarshaller.unmarshal(request, response)
|
