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
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
|
from __future__ import annotations
from dataclasses import dataclass, fields
from typing import Any
from xsdata.exceptions import ClientValueError
from xsdata.formats.dataclass.context import XmlContext
from xsdata.formats.dataclass.parsers import DictDecoder, XmlParser
from xsdata.formats.dataclass.serializers import XmlSerializer
from xsdata.formats.dataclass.transports import DefaultTransport, Transport
@dataclass(frozen=True)
class Config:
"""Service configuration class.
Attributes:
style: The binding style
location: The service endpoint url
transport: The transport namespace
soap_action: The soap action
input: The input class
output: The output class
"""
style: str
location: str
transport: str
soap_action: str
input: type
output: type
encoding: str | None = None
@classmethod
def from_service(cls, obj: Any, **kwargs: Any) -> Config:
"""Instantiate from a generated service class.
Args:
obj: The service class
**kwargs: Override the service class properties
style: The binding style
location: The service endpoint url
transport: The transport namespace
soap_action: The soap action
input: The input class
output: The output class
Returns:
A new config instance.
"""
params = {
f.name: kwargs[f.name] if f.name in kwargs else getattr(obj, f.name, None)
for f in fields(cls)
}
return cls(**params) # type: ignore
class TransportTypes:
"""Transport types."""
SOAP = "http://schemas.xmlsoap.org/soap/http"
class Client:
"""A wsdl client.
Args:
config: The service config instance
transport: The transport instance
parser: The xml parser instance
serializer: The xml serializer instance
"""
__slots__ = "config", "parser", "serializer", "transport"
def __init__(
self,
config: Config,
transport: Transport | None = None,
parser: XmlParser | None = None,
serializer: XmlSerializer | None = None,
):
"""Initialize the client."""
self.config = config
self.transport = transport or DefaultTransport()
if not serializer and not parser:
context = XmlContext()
serializer = XmlSerializer(context=context)
parser = XmlParser(context=context)
elif not serializer:
assert parser is not None
serializer = XmlSerializer(context=parser.context)
else:
assert serializer is not None
parser = XmlParser(context=serializer.context)
self.parser = parser
self.serializer = serializer
@classmethod
def from_service(cls, obj: type, **kwargs: Any) -> Client:
"""Instantiate client from a service class.
Args:
obj: The service class
**kwargs: Override the service class properties
style: The binding style
location: The service endpoint url
transport: The transport namespace
soap_action: The soap action
input: The input class
output: The output class
Returns:
A new client instance.
"""
return cls(config=Config.from_service(obj, **kwargs))
def send(self, obj: Any, headers: dict | None = None) -> Any:
"""Build and send a request for the input object.
```py
params = {"body": {"add": {"int_a": 3, "int_b": 4}}}
res = client.send(params)
```
Is equivalent with:
```py
req = CalculatorSoapAddInput(body=CalculatorSoapAddInput.Body(add=Add(3, 4)))
res = client.send(req)
```
Args:
obj: The request model instance or a pure dictionary
headers: Additional headers to pass to the transport
Returns:
The response model instance.
"""
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 the request headers.
It merges the custom user headers with the necessary headers
to accommodate the service class configuration.
Raises:
ClientValueError: If the service transport type is not supported.
"""
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) -> str | bytes:
"""Prepare and serialize the payload to be sent.
If the obj is a pure dictionary, it will be converted
first to a request model instance.
Args:
obj: The request model instance or a pure dictionary
Returns:
The serialized request body content as string or bytes.
Raises:
ClientValueError: If the config input type doesn't match the given object.
"""
if isinstance(obj, dict):
decoder = DictDecoder(context=self.serializer.context)
obj = decoder.decode(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
|
