/usr/lib/python3/dist-packages/zeep/plugins.py is in python3-zeep 2.5.0-1.
This file is owned by root:root, with mode 0o644.
The actual contents of the file can be viewed below.
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 | from collections import deque
class Plugin(object):
"""Base plugin"""
def ingress(self, envelope, http_headers, operation):
"""Override to update the envelope or http headers when receiving a
message.
:param envelope: The envelope as XML node
:param http_headers: Dict with the HTTP headers
"""
return envelope, http_headers
def egress(self, envelope, http_headers, operation, binding_options):
"""Override to update the envelope or http headers when sending a
message.
:param envelope: The envelope as XML node
:param http_headers: Dict with the HTTP headers
:param operation: The associated Operation instance
:param binding_options: Binding specific options for the operation
"""
return envelope, http_headers
def apply_egress(client, envelope, http_headers, operation, binding_options):
for plugin in client.plugins:
result = plugin.egress(
envelope, http_headers, operation, binding_options)
if result is not None:
envelope, http_headers = result
return envelope, http_headers
def apply_ingress(client, envelope, http_headers, operation):
for plugin in client.plugins:
result = plugin.ingress(envelope, http_headers, operation)
if result is not None:
envelope, http_headers = result
return envelope, http_headers
class HistoryPlugin(object):
def __init__(self, maxlen=1):
self._buffer = deque([], maxlen)
@property
def last_sent(self):
last_tx = self._buffer[-1]
if last_tx:
return last_tx['sent']
@property
def last_received(self):
last_tx = self._buffer[-1]
if last_tx:
return last_tx['received']
def ingress(self, envelope, http_headers, operation):
last_tx = self._buffer[-1]
last_tx['received'] = {
'envelope': envelope,
'http_headers': http_headers,
}
def egress(self, envelope, http_headers, operation, binding_options):
self._buffer.append({
'received': None,
'sent': {
'envelope': envelope,
'http_headers': http_headers,
},
})
|