libvirt-override.py

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