bt2: make it work for Python 3.4
[babeltrace.git] / src / bindings / python / bt2 / bt2 / query_executor.py
1 # The MIT License (MIT)
2 #
3 # Copyright (c) 2017 Philippe Proulx <pproulx@efficios.com>
4 #
5 # Permission is hereby granted, free of charge, to any person obtaining a copy
6 # of this software and associated documentation files (the "Software"), to deal
7 # in the Software without restriction, including without limitation the rights
8 # to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
9 # copies of the Software, and to permit persons to whom the Software is
10 # furnished to do so, subject to the following conditions:
11 #
12 # The above copyright notice and this permission notice shall be included in
13 # all copies or substantial portions of the Software.
14 #
15 # THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
16 # IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
17 # FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
18 # AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
19 # LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
20 # OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
21 # THE SOFTWARE.
22
23 from bt2 import native_bt, object, utils
24 from bt2 import interrupter as bt2_interrupter
25 from bt2 import value as bt2_value
26 import bt2
27
28
29 def _bt2_component():
30 from bt2 import component as bt2_component
31
32 return bt2_component
33
34
35 class _QueryExecutorCommon:
36 @property
37 def _common_ptr(self):
38 return self._as_query_executor_ptr()
39
40 @property
41 def is_interrupted(self):
42 is_interrupted = native_bt.query_executor_is_interrupted(self._common_ptr)
43 return bool(is_interrupted)
44
45 @property
46 def logging_level(self):
47 return native_bt.query_executor_get_logging_level(self._common_ptr)
48
49
50 class QueryExecutor(object._SharedObject, _QueryExecutorCommon):
51 _get_ref = staticmethod(native_bt.query_executor_get_ref)
52 _put_ref = staticmethod(native_bt.query_executor_put_ref)
53
54 def _as_query_executor_ptr(self):
55 return self._ptr
56
57 def __init__(self, component_class, object, params=None, method_obj=None):
58 if not isinstance(component_class, _bt2_component()._ComponentClassConst):
59 err = False
60
61 try:
62 if not issubclass(component_class, _bt2_component()._UserComponent):
63 err = True
64 except TypeError:
65 err = True
66
67 if err:
68 o = component_class
69 raise TypeError("'{}' is not a component class object".format(o))
70
71 utils._check_str(object)
72
73 if params is None:
74 params_ptr = native_bt.value_null
75 else:
76 params = bt2.create_value(params)
77 params_ptr = params._ptr
78
79 cc_ptr = component_class._bt_component_class_ptr()
80 assert cc_ptr is not None
81
82 if method_obj is not None and not native_bt.bt2_is_python_component_class(
83 cc_ptr
84 ):
85 raise ValueError(
86 'cannot pass a Python object to a non-Python component class'
87 )
88
89 ptr = native_bt.bt2_query_executor_create(
90 cc_ptr, object, params_ptr, method_obj
91 )
92
93 if ptr is None:
94 raise bt2._MemoryError('cannot create query executor object')
95
96 super().__init__(ptr)
97
98 # Keep a reference of `method_obj` as the native query executor
99 # does not have any. This ensures that, when this object's
100 # query() method is called, the Python object still exists.
101 self._method_obj = method_obj
102
103 def add_interrupter(self, interrupter):
104 utils._check_type(interrupter, bt2_interrupter.Interrupter)
105 native_bt.query_executor_add_interrupter(self._ptr, interrupter._ptr)
106
107 def interrupt(self):
108 native_bt.query_executor_interrupt(self._ptr)
109
110 def _set_logging_level(self, log_level):
111 utils._check_log_level(log_level)
112 status = native_bt.query_executor_set_logging_level(self._ptr, log_level)
113 utils._handle_func_status(status, "cannot set query executor's logging level")
114
115 logging_level = property(
116 fget=_QueryExecutorCommon.logging_level, fset=_set_logging_level
117 )
118
119 @property
120 def is_interrupted(self):
121 is_interrupted = native_bt.query_executor_is_interrupted(self._ptr)
122 return bool(is_interrupted)
123
124 def query(self):
125 status, result_ptr = native_bt.query_executor_query(self._ptr)
126 utils._handle_func_status(status, 'cannot query component class')
127 assert result_ptr is not None
128 return bt2_value._create_from_const_ptr(result_ptr)
129
130
131 class _PrivateQueryExecutor(_QueryExecutorCommon):
132 def __init__(self, ptr):
133 self._ptr = ptr
134
135 def _check_validity(self):
136 if self._ptr is None:
137 raise RuntimeError('this object is not valid anymore')
138
139 def _as_query_executor_ptr(self):
140 self._check_validity()
141 return native_bt.private_query_executor_as_query_executor_const(self._ptr)
142
143 def _invalidate(self):
144 self._ptr = None
This page took 0.032345 seconds and 4 git commands to generate.