1# 2# Manually written part of python bindings for libvirt 3# 4 5# On cygwin, the DLL is called cygvirtmod.dll 6try: 7 import libvirtmod # type: ignore 8except ImportError as lib_e: 9 try: 10 import cygvirtmod as libvirtmod # type: ignore 11 except ImportError as cyg_e: 12 if "No module named" in str(cyg_e): 13 raise lib_e 14 15from types import TracebackType 16from typing import Any, Callable, Dict, List, Optional, overload, Tuple, Type, TypeVar, Union 17_T = TypeVar('_T') 18_EventCB = Callable[[int, int, int, _T], None] 19_EventAddHandleFunc = Callable[[int, int, _EventCB, _T], int] 20_EventUpdateHandleFunc = Callable[[int, int], None] 21_EventRemoveHandleFunc = Callable[[int], int] 22_TimerCB = Callable[[int, _T], None] 23_EventAddTimeoutFunc = Callable[[int, _TimerCB, _T], int] 24_EventUpdateTimeoutFunc = Callable[[int, int], None] 25_EventRemoveTimeoutFunc = Callable[[int], int] 26_DomainCB = Callable[['virConnect', 'virDomain', int, int, _T], Optional[int]] 27_BlkioParameter = Dict[str, Any] 28_MemoryParameter = Dict[str, Any] 29_SchedParameter = Dict[str, Any] 30_TypedParameter = Dict[str, Any] 31 32 33# The root of all libvirt errors. 34class libvirtError(Exception): 35 def __init__(self, defmsg: str) -> None: 36 37 # Never call virConnGetLastError(). 38 # virGetLastError() is now thread local 39 err = libvirtmod.virGetLastError() # type: Optional[Tuple[int, int, str, int, str, Optional[str], Optional[str], int, int]] 40 if err is None: 41 msg = defmsg 42 else: 43 msg = err[2] 44 45 Exception.__init__(self, msg) 46 47 self.err = err 48 49 def get_error_code(self) -> Optional[int]: 50 if self.err is None: 51 return None 52 return self.err[0] 53 54 def get_error_domain(self) -> Optional[int]: 55 if self.err is None: 56 return None 57 return self.err[1] 58 59 def get_error_message(self) -> Optional[str]: 60 if self.err is None: 61 return None 62 return self.err[2] 63 64 def get_error_level(self) -> Optional[int]: 65 if self.err is None: 66 return None 67 return self.err[3] 68 69 def get_str1(self) -> Optional[str]: 70 if self.err is None: 71 return None 72 return self.err[4] 73 74 def get_str2(self) -> Optional[str]: 75 if self.err is None: 76 return None 77 return self.err[5] 78 79 def get_str3(self) -> Optional[str]: 80 if self.err is None: 81 return None 82 return self.err[6] 83 84 def get_int1(self) -> Optional[int]: 85 if self.err is None: 86 return None 87 return self.err[7] 88 89 def get_int2(self) -> Optional[int]: 90 if self.err is None: 91 return None 92 return self.err[8] 93 94 95# 96# register the libvirt global error handler 97# 98def registerErrorHandler(f: Callable[[_T, List], None], ctx: _T) -> int: 99 """Register a Python function for error reporting. 100 The function is called back as f(ctx, error), with error 101 being a list of information about the error being raised. 102 Returns 1 in case of success.""" 103 return libvirtmod.virRegisterErrorHandler(f, ctx) 104 105 106def openAuth(uri: str, auth: List, flags: int = 0) -> 'virConnect': 107 # TODO: The C code rquires a List and there is not *Mutable*Tuple for a better description such as 108 # auth: Tuple[List[int], Callable[[List[MutableTuple[int, str, str, str, Any]], _T], int], _T] 109 """ 110 This function should be called first to get a connection to the 111 Hypervisor. If necessary, authentication will be performed fetching 112 credentials via the callback. 113 114 See :py:func:`open` for notes about environment variables which can 115 have an effect on opening drivers and freeing the connection resources. 116 117 :param str uri: (Optional) connection URI, see https://libvirt.org/uri.html 118 :param auth: a list that contains 3 items: 119 - a list of supported credential types 120 - a callable that takes 2 arguments (credentials, user-data) and returns 0 on succcess and -1 on errors. 121 The credentials argument is a list of credentials that libvirt (actually 122 the ESX driver) would like to request. An element of this list is itself a 123 list containing 5 items (4 inputs, 1 output): 124 - the credential type, e.g. :py:const:`libvirt.VIR_CRED_AUTHNAME` 125 - a prompt to be displayed to the user 126 - a challenge, the ESX driver sets this to the hostname to allow automatic 127 distinction between requests for ESX and vCenter credentials 128 - a default result for the request 129 - a place to store the actual result for the request 130 - user data that will be passed to the callable as second argument 131 :param int flags: bitwise-OR of virConnectFlags 132 :returns: a :py:class:`virConnect` instance on success. 133 :raises libvirtError: on errors. 134 """ 135 ret = libvirtmod.virConnectOpenAuth(uri, auth, flags) 136 if ret is None: 137 raise libvirtError('virConnectOpenAuth() failed') 138 return virConnect(_obj=ret) 139 140 141# 142# Return library version. 143# 144def getVersion(name: Optional[str] = None) -> int: 145 """If no name parameter is passed (or name is None) then the 146 version of the libvirt library is returned as an integer. 147 148 If a name is passed and it refers to a driver linked to the 149 libvirt library, then this returns a tuple of (library version, 150 driver version). 151 152 If the name passed refers to a non-existent driver, then you 153 will get the exception 'no support for hypervisor'. 154 155 Versions numbers are integers: 1000000*major + 1000*minor + release.""" 156 if name is None: 157 ret = libvirtmod.virGetVersion() 158 else: 159 ret = libvirtmod.virGetVersion(name) 160 if ret is None: 161 raise libvirtError("virGetVersion() failed") 162 return ret 163 164 165# 166# Invoke an EventHandle callback 167# 168@overload 169def _eventInvokeHandleCallback(watch: int, fd: int, event: int, opaque: Tuple[_EventCB, _T], opaquecompat: None = None) -> None: ... # noqa E704 170@overload # noqa F811 171def _eventInvokeHandleCallback(watch: int, fd: int, event: int, opaque: _EventCB, opaquecompat: _T = None) -> None: ... # noqa E704 172def _eventInvokeHandleCallback(watch: int, fd: int, event: int, opaque: Union[Tuple[_EventCB, _T], _EventCB], opaquecompat: Optional[_T] = None) -> None: # noqa F811 173 """ 174 Invoke the Event Impl Handle Callback in C 175 """ 176 # libvirt 0.9.2 and earlier required custom event loops to know 177 # that opaque=(cb, original_opaque) and pass the values individually 178 # to this wrapper. This should handle the back compat case, and make 179 # future invocations match the virEventHandleCallback prototype 180 if opaquecompat: 181 callback = opaque 182 opaque_ = opaquecompat 183 else: 184 assert isinstance(opaque, tuple) 185 callback = opaque[0] 186 opaque_ = opaque[1] 187 188 libvirtmod.virEventInvokeHandleCallback(watch, fd, event, callback, opaque_) 189 190 191# 192# Invoke an EventTimeout callback 193# 194def _eventInvokeTimeoutCallback(timer: int, opaque: Union[Tuple[_TimerCB, _T], _TimerCB], opaquecompat: Optional[_T] = None) -> None: 195 """ 196 Invoke the Event Impl Timeout Callback in C 197 """ 198 # libvirt 0.9.2 and earlier required custom event loops to know 199 # that opaque=(cb, original_opaque) and pass the values individually 200 # to this wrapper. This should handle the back compat case, and make 201 # future invocations match the virEventTimeoutCallback prototype 202 if opaquecompat: 203 callback = opaque 204 opaque_ = opaquecompat 205 else: 206 assert isinstance(opaque, tuple) 207 callback = opaque[0] 208 opaque_ = opaque[1] 209 210 libvirtmod.virEventInvokeTimeoutCallback(timer, callback, opaque_) 211 212 213def _dispatchEventHandleCallback(watch: int, fd: int, events: int, cbData: Dict[str, Any]) -> int: 214 cb = cbData["cb"] 215 opaque = cbData["opaque"] 216 217 cb(watch, fd, events, opaque) 218 return 0 219 220 221def _dispatchEventTimeoutCallback(timer: int, cbData: Dict[str, Any]) -> int: 222 cb = cbData["cb"] 223 opaque = cbData["opaque"] 224 225 cb(timer, opaque) 226 return 0 227 228 229def virEventAddHandle(fd: int, events: int, cb: _EventCB, opaque: _T) -> int: 230 """ 231 register a callback for monitoring file handle events 232 233 @fd: file handle to monitor for events 234 @events: bitset of events to watch from virEventHandleType constants 235 @cb: callback to invoke when an event occurs 236 @opaque: user data to pass to callback 237 238 Example callback prototype is: 239 def cb(watch, # int id of the handle 240 fd, # int file descriptor the event occurred on 241 events, # int bitmap of events that have occurred 242 opaque): # opaque data passed to eventAddHandle 243 """ 244 cbData = {"cb": cb, "opaque": opaque} 245 ret = libvirtmod.virEventAddHandle(fd, events, cbData) 246 if ret == -1: 247 raise libvirtError('virEventAddHandle() failed') 248 return ret 249 250 251def virEventAddTimeout(timeout: int, cb: _TimerCB, opaque: _T) -> int: 252 """ 253 register a callback for a timer event 254 255 @timeout: time between events in milliseconds 256 @cb: callback to invoke when an event occurs 257 @opaque: user data to pass to callback 258 259 Setting timeout to -1 will disable the timer. Setting the timeout 260 to zero will cause it to fire on every event loop iteration. 261 262 Example callback prototype is: 263 def cb(timer, # int id of the timer 264 opaque): # opaque data passed to eventAddTimeout 265 """ 266 cbData = {"cb": cb, "opaque": opaque} 267 ret = libvirtmod.virEventAddTimeout(timeout, cbData) 268 if ret == -1: 269 raise libvirtError('virEventAddTimeout() failed') 270 return ret 271 272 273# 274# a caller for the ff callbacks for custom event loop implementations 275# 276 277def virEventInvokeFreeCallback(opaque: Any) -> None: 278 """ 279 Execute callback which frees the opaque buffer 280 281 @opaque: the opaque object passed to addHandle or addTimeout 282 283 WARNING: This function should not be called from any call by libvirt's 284 core. It will most probably cause deadlock in C-level libvirt code. 285 Instead it should be scheduled and called from implementation's stack. 286 287 See https://libvirt.org/html/libvirt-libvirt-event.html#virEventAddHandleFunc 288 for more information. 289 290 This function is not dependent on any event loop implementation. 291 """ 292 293 libvirtmod.virEventInvokeFreeCallback(opaque[2], opaque[1]) 294