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
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
|
from dataclasses import dataclass, field
from typing import Any, Dict, NamedTuple, Optional, Type
from xsdata.exceptions import ClientValueError
from xsdata.formats.dataclass.parsers import XmlParser
from xsdata.formats.dataclass.parsers.json import DictConverter
from xsdata.formats.dataclass.serializers import XmlSerializer
from xsdata.formats.dataclass.transports import DefaultTransport, Transport
class Config(NamedTuple):
"""
Service configuration class.
:param style: binding style
:param location: service endpoint url
:param transport: transport namespace
:param soap_action: soap action
:param input: input object type
:param output: output object type
"""
style: str
location: str
transport: str
soap_action: str
input: Type
output: Type
encoding: Optional[str] = None
@classmethod
def from_service(cls, obj: Any, **kwargs: Any) -> "Config":
"""Instantiate from a generated service class."""
params = {
key: kwargs[key] if key in kwargs else getattr(obj, key, None)
for key in cls._fields
}
return cls(**params)
class TransportTypes:
SOAP = "http://schemas.xmlsoap.org/soap/http"
@dataclass
class Client:
"""
:param config: service configuration
:param transport: transport instance to handle requests
:param parser: xml parser instance to handle xml response parsing
:param serializer: xml serializer instance to handle xml response parsing
"""
config: Config
transport: Transport = field(default_factory=DefaultTransport)
parser: XmlParser = field(default_factory=XmlParser)
serializer: XmlSerializer = field(default_factory=XmlSerializer)
dict_converter: DictConverter = field(init=False, default_factory=DictConverter)
@classmethod
def from_service(cls, obj: Type, **kwargs: str) -> "Client":
"""Instantiate client from a service definition."""
return cls(config=Config.from_service(obj, **kwargs))
def send(self, obj: Any, headers: Optional[Dict] = None) -> Any:
"""
Send a request and parse the response according to the service
configuration.
The input object can be a dictionary, or the input type instance directly
>>> params = {"body": {"add": {"int_a": 3, "int_b": 4}}}
>>> res = client.send(params)
Is equivalent with:
>>> req = CalculatorSoapAddInput(
>>> body=CalculatorSoapAddInput.Body(add=Add(3, 4)))
>>> res = client.send(req)
:param obj: a params dictionary or the input type instance
:param headers: a dictionary of any additional headers.
"""
data = self.prepare_payload(obj)
headers = self.prepare_headers(headers or {})
response = self.transport.post(self.config.location, data=data, headers=headers)
return self.parser.from_bytes(response, self.config.output)
def prepare_headers(self, headers: Dict) -> Dict:
"""
Prepare request headers according to the service configuration.
Don't mutate input headers dictionary.
:raises ClientValueError: If the service transport type is
unsupported.
"""
result = headers.copy()
if self.config.transport == TransportTypes.SOAP:
result["content-type"] = "text/xml"
if self.config.soap_action:
result["SOAPAction"] = self.config.soap_action
else:
raise ClientValueError(
f"Unsupported binding transport: `{self.config.transport}`"
)
return result
def prepare_payload(self, obj: Any) -> Any:
"""
Prepare and serialize payload to be sent.
:raises ClientValueError: If the config input type doesn't match
the given input.
"""
if isinstance(obj, Dict):
obj = self.dict_converter.convert(obj, self.config.input)
if not isinstance(obj, self.config.input):
raise ClientValueError(
f"Invalid input service type, "
f"expected `{self.config.input.__name__}` "
f"got `{type(obj).__name__}`"
)
result = self.serializer.render(obj)
if self.config.encoding:
return result.encode(self.config.encoding)
return result
|
