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
|
import json
from types import SimpleNamespace
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},
},
)
|
