k3nny/TISbackup
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix iniparse
lint / docker (push) Successful in 9m14s
Details

Browse Source 
fix code passing ruff linter
pre-commit ruff
pre-commit ruff format
This commit is contained in:
k3nny
2024-11-29 23:45:40 +01:00
parent aa8a68aa80
commit 737f9bea38
27 changed files with 2375 additions and 2016 deletions
Download Patch File Download Diff File Expand all files Collapse all files
libtisbackup/XenAPI.py
+48 -49
View File
@@ -61,10 +61,11 @@ import sys
import six.moves.http_client as httplib
import six.moves.xmlrpc_client as xmlrpclib
translation = gettext.translation('xen-xm', fallback = True)
translation = gettext.translation("xen-xm", fallback=True)
API_VERSION_1_1 = "1.1"
API_VERSION_1_2 = "1.2"
API_VERSION_1_1 = '1.1'
API_VERSION_1_2 = '1.2'
class Failure(Exception):
def __init__(self, details):
@@ -79,41 +80,48 @@ class Failure(Exception):
return msg
def _details_map(self):
return dict([(str(i), self.details[i])
for i in range(len(self.details))])
return dict([(str(i), self.details[i]) for i in range(len(self.details))])
# Just a "constant" that we use to decide whether to retry the RPC
_RECONNECT_AND_RETRY = object()
class UDSHTTPConnection(httplib.HTTPConnection):
"""HTTPConnection subclass to allow HTTP over Unix domain sockets. """
"""HTTPConnection subclass to allow HTTP over Unix domain sockets."""
def connect(self):
path = self.host.replace("_", "/")
self.sock = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
self.sock.connect(path)
class UDSHTTP(httplib.HTTPConnection):
_connection_class = UDSHTTPConnection
class UDSTransport(xmlrpclib.Transport):
def __init__(self, use_datetime=0):
self._use_datetime = use_datetime
self._extra_headers=[]
self._extra_headers = []
self._connection = (None, None)
def add_extra_header(self, key, value):
self._extra_headers += [ (key,value) ]
self._extra_headers += [(key, value)]
def make_connection(self, host):
# Python 2.4 compatibility
if sys.version_info[0] <= 2 and sys.version_info[1] < 7:
return UDSHTTP(host)
else:
return UDSHTTPConnection(host)
def send_request(self, connection, handler, request_body):
connection.putrequest("POST", handler)
for key, value in self._extra_headers:
connection.putheader(key, value)
class Session(xmlrpclib.ServerProxy):
"""A server proxy and session manager for communicating with xapi using
the Xen-API.
@@ -126,32 +134,27 @@ class Session(xmlrpclib.ServerProxy):
session.xenapi.session.logout()
"""
def __init__(self, uri, transport=None, encoding=None, verbose=0,
allow_none=1, ignore_ssl=False):
def __init__(self, uri, transport=None, encoding=None, verbose=0, allow_none=1, ignore_ssl=False):
# Fix for CA-172901 (+ Python 2.4 compatibility)
# Fix for context=ctx ( < Python 2.7.9 compatibility)
if not (sys.version_info[0] <= 2 and sys.version_info[1] <= 7 and sys.version_info[2] <= 9 ) \
and ignore_ssl:
if not (sys.version_info[0] <= 2 and sys.version_info[1] <= 7 and sys.version_info[2] <= 9) and ignore_ssl:
import ssl
ctx = ssl._create_unverified_context()
xmlrpclib.ServerProxy.__init__(self, uri, transport, encoding,
verbose, allow_none, context=ctx)
xmlrpclib.ServerProxy.__init__(self, uri, transport, encoding, verbose, allow_none, context=ctx)
else:
xmlrpclib.ServerProxy.__init__(self, uri, transport, encoding,
verbose, allow_none)
xmlrpclib.ServerProxy.__init__(self, uri, transport, encoding, verbose, allow_none)
self.transport = transport
self._session = None
self.last_login_method = None
self.last_login_params = None
self.API_version = API_VERSION_1_1
def xenapi_request(self, methodname, params):
if methodname.startswith('login'):
if methodname.startswith("login"):
self._login(methodname, params)
return None
elif methodname == 'logout' or methodname == 'session.logout':
elif methodname == "logout" or methodname == "session.logout":
self._logout()
return None
else:
@@ -162,29 +165,25 @@ class Session(xmlrpclib.ServerProxy):
if result is _RECONNECT_AND_RETRY:
retry_count += 1
if self.last_login_method:
self._login(self.last_login_method,
self.last_login_params)
self._login(self.last_login_method, self.last_login_params)
else:
raise xmlrpclib.Fault(401, 'You must log in')
raise xmlrpclib.Fault(401, "You must log in")
else:
return result
raise xmlrpclib.Fault(
500, 'Tried 3 times to get a valid session, but failed')
raise xmlrpclib.Fault(500, "Tried 3 times to get a valid session, but failed")
def _login(self, method, params):
try:
result = _parse_result(
getattr(self, 'session.%s' % method)(*params))
result = _parse_result(getattr(self, "session.%s" % method)(*params))
if result is _RECONNECT_AND_RETRY:
raise xmlrpclib.Fault(
500, 'Received SESSION_INVALID when logging in')
raise xmlrpclib.Fault(500, "Received SESSION_INVALID when logging in")
self._session = result
self.last_login_method = method
self.last_login_params = params
self.API_version = self._get_api_version()
except socket.error as e:
if e.errno == socket.errno.ETIMEDOUT:
raise xmlrpclib.Fault(504, 'The connection timed out')
raise xmlrpclib.Fault(504, "The connection timed out")
else:
raise e
@@ -205,41 +204,41 @@ class Session(xmlrpclib.ServerProxy):
host = self.xenapi.pool.get_master(pool)
major = self.xenapi.host.get_API_version_major(host)
minor = self.xenapi.host.get_API_version_minor(host)
return "%s.%s"%(major,minor)
return "%s.%s" % (major, minor)
def __getattr__(self, name):
if name == 'handle':
if name == "handle":
return self._session
elif name == 'xenapi':
elif name == "xenapi":
return _Dispatcher(self.API_version, self.xenapi_request, None)
elif name.startswith('login') or name.startswith('slave_local'):
elif name.startswith("login") or name.startswith("slave_local"):
return lambda *params: self._login(name, params)
elif name == 'logout':
elif name == "logout":
return _Dispatcher(self.API_version, self.xenapi_request, "logout")
else:
return xmlrpclib.ServerProxy.__getattr__(self, name)
def xapi_local():
return Session("http://_var_lib_xcp_xapi/", transport=UDSTransport())
def _parse_result(result):
if type(result) != dict or 'Status' not in result:
raise xmlrpclib.Fault(500, 'Missing Status in response from server' + result)
if result['Status'] == 'Success':
if 'Value' in result:
return result['Value']
if not isinstance(type(result), dict) or "Status" not in result:
raise xmlrpclib.Fault(500, "Missing Status in response from server" + result)
if result["Status"] == "Success":
if "Value" in result:
return result["Value"]
else:
raise xmlrpclib.Fault(500,
'Missing Value in response from server')
raise xmlrpclib.Fault(500, "Missing Value in response from server")
else:
if 'ErrorDescription' in result:
if result['ErrorDescription'][0] == 'SESSION_INVALID':
if "ErrorDescription" in result:
if result["ErrorDescription"][0] == "SESSION_INVALID":
return _RECONNECT_AND_RETRY
else:
raise Failure(result['ErrorDescription'])
raise Failure(result["ErrorDescription"])
else:
raise xmlrpclib.Fault(
500, 'Missing ErrorDescription in response from server')
raise xmlrpclib.Fault(500, "Missing ErrorDescription in response from server")
# Based upon _Method from xmlrpclib.
@@ -251,9 +250,9 @@ class _Dispatcher:
def __repr__(self):
if self.__name:
return '<XenAPI._Dispatcher for %s>' % self.__name
return "<XenAPI._Dispatcher for %s>" % self.__name
else:
return '<XenAPI._Dispatcher>'
return "<XenAPI._Dispatcher>"
def __getattr__(self, name):
if self.__name is None: