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
|
from __future__ import print_function
from __future__ import unicode_literals
import json
try:
from types import SimpleNamespace
ConnectionError = ConnectionError
BrokenPipeError = BrokenPipeError
except: # Python 2
from argparse import Namespace as SimpleNamespace
class ConnectionError(OSError):
pass
class BrokenPipeError(ConnectionError):
pass
class VarlinkEncoder(json.JSONEncoder):
"""The Encoder used to encode JSON"""
def default(self, o):
if isinstance(o, set):
return dict.fromkeys(o, {})
if isinstance(o, SimpleNamespace):
return o.__dict__
if isinstance(o, VarlinkError):
return o.as_dict()
return json.JSONEncoder.default(self, o)
class VarlinkError(Exception):
"""The base class for varlink error exceptions"""
@classmethod
def new(cls, message, namespaced=False):
if message['error'] == 'org.varlink.service.InterfaceNotFound':
return InterfaceNotFound.new(message, namespaced)
elif message['error'] == 'org.varlink.service.InvalidParameter':
return InvalidParameter.new(message, namespaced)
elif message['error'] == 'org.varlink.service.MethodNotImplemented':
return MethodNotImplemented.new(message, namespaced)
elif message['error'] == 'org.varlink.service.MethodNotImplemented':
return MethodNotImplemented.new(message, namespaced)
else:
return cls(message, namespaced)
def __init__(self, message, namespaced=False):
if not namespaced and not isinstance(message, dict):
raise TypeError
# normalize to dictionary
Exception.__init__(self, json.loads(json.dumps(message, cls=VarlinkEncoder)))
def error(self):
"""returns the exception varlink error name"""
return self.args[0].get('error')
def parameters(self, namespaced=False):
"""returns the exception varlink error parameters"""
if namespaced:
return json.loads(json.dumps(self.args[0]['parameters']), object_hook=lambda d: SimpleNamespace(**d))
else:
return self.args[0].get('parameters')
def as_dict(self):
return self.args[0]
class InterfaceNotFound(VarlinkError):
"""The standardized varlink InterfaceNotFound error as a python exception"""
@classmethod
def new(cls, message, namespaced=False):
return cls(namespaced and message['parameters'].interface or message['parameters'].get('interface', None))
def __init__(self, interface):
VarlinkError.__init__(self, {'error': 'org.varlink.service.InterfaceNotFound',
'parameters': {'interface': interface}})
class MethodNotFound(VarlinkError):
"""The standardized varlink MethodNotFound error as a python exception"""
@classmethod
def new(cls, message, namespaced=False):
return cls(namespaced and message['parameters'].method or message['parameters'].get('method', None))
def __init__(self, method):
VarlinkError.__init__(self, {'error': 'org.varlink.service.MethodNotFound', 'parameters': {'method': method}})
class MethodNotImplemented(VarlinkError):
"""The standardized varlink MethodNotImplemented error as a python exception"""
@classmethod
def new(cls, message, namespaced=False):
return cls(namespaced and message['parameters'].method or message['parameters'].get('method', None))
def __init__(self, method):
VarlinkError.__init__(self,
{'error': 'org.varlink.service.MethodNotImplemented', 'parameters': {'method': method}})
class InvalidParameter(VarlinkError):
"""The standardized varlink InvalidParameter error as a python exception"""
@classmethod
def new(cls, message, namespaced=False):
return cls(namespaced and message['parameters'].parameter or message['parameters'].get('parameter', None))
def __init__(self, name):
VarlinkError.__init__(self,
{'error': 'org.varlink.service.InvalidParameter', 'parameters': {'parameter': name}})
|
