Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Jump to bottom

Cleanup core.py #122

Merged
merged 4 commits into from Nov 20, 2022
Merged

Cleanup core.py #122

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
cd92589
Cleanup core.py
jhollowe Nov 19, 2022
bcec27a
Use f-strings
jhollowe Nov 19, 2022
d5abb5c
Add start of docstrings
jhollowe Nov 20, 2022
a401b91
Merge branch 'develop' into cleanup-core
jhollowe Nov 20, 2022
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Jump to
Jump to file
Failed to load files.
Diff view
Diff view
8 changes: 4 additions & 4 deletions proxmoxer/backends/command_base.py
View file
Open in desktop
Expand Up @@ -74,7 +74,7 @@ def request(self, method, url, data=None, params=None, headers=None):
tmp_filename = ""
if url.endswith("upload"):
# copy file to temporary location on proxmox host
tmp_filename, tmp_err = self._exec(
tmp_filename, _ = self._exec(
[
"python3",
"-c",
Expand All @@ -86,9 +86,9 @@ def request(self, method, url, data=None, params=None, headers=None):
data["filename"] = data["filename"].name
data["tmpfilename"] = tmp_filename

command = ["{0}sh".format(self.service), cmd, url]
command = [f"{self.service}sh", cmd, url]
# convert the options dict into a 2-tuple with the key formatted as a flag
option_pairs = [("-{0}".format(k), str(v)) for k, v in chain(data.items(), params.items())]
option_pairs = [(f"-{k}", str(v)) for k, v in chain(data.items(), params.items())]
# add back in all the command arguments as their own pairs
if data_command is not None:
if isinstance(data_command, list):
Expand All @@ -99,7 +99,7 @@ def request(self, method, url, data=None, params=None, headers=None):
option_pairs.append(("-command", arg))
# expand the list of 2-tuples into a flat list
options = [val for pair in option_pairs for val in pair]
additional_options = SERVICES[self.service.upper()].get("ssh_additional_options", [])
additional_options = SERVICES[self.service.upper()].get("cli_additional_options", [])
full_cmd = command + options + additional_options

if self.sudo:
Expand Down
8 changes: 4 additions & 4 deletions proxmoxer/backends/https.py
View file
Open in desktop
Expand Up @@ -112,7 +112,7 @@ def get_tokens(self):
def __call__(self, req):
# refresh ticket if older than `renew_age`
if (get_time() - self.birth_time) >= self.renew_age:
logger.debug("refreshing ticket (age %d)", (get_time() - self.birth_time))
logger.debug(f"refreshing ticket (age {get_time() - self.birth_time})")
self._get_new_tokens()

# only attach CSRF token if needed (reduce interception risk)
Expand Down Expand Up @@ -282,7 +282,7 @@ def __init__(
if "]:" in host:
host, host_port = host.rsplit(":", 1)
else:
host = "[{0}]".format(host)
host = f"[{host}]"
elif ":" in host:
host, host_port = host.split(":")
port = host_port if host_port.isdigit() else port
Expand All @@ -293,9 +293,9 @@ def __init__(

self.mode = mode
if path_prefix is not None:
self.base_url = "https://{0}:{1}/{2}/api2/{3}".format(host, port, path_prefix, mode)
self.base_url = f"https://{host}:{port}/{path_prefix}/api2/{mode}"
else:
self.base_url = "https://{0}:{1}/api2/{2}".format(host, port, mode)
self.base_url = f"https://{host}:{port}/api2/{mode}"

if token_name is not None:
if "token" not in SERVICES[service]["supported_https_auths"]:
Expand Down
54 changes: 29 additions & 25 deletions proxmoxer/core.py
View file
Open in desktop
Expand Up @@ -7,20 +7,8 @@
import importlib
import logging
import posixpath

# Python 3 compatibility:
try:
import httplib
except ImportError: # py3
from http import client as httplib
try:
import urlparse
except ImportError: # py3
from urllib import parse as urlparse
try:
basestring
except NameError: # py3
basestring = (bytes, str)
from http import client as httplib
from urllib import parse as urlparse

logger = logging.getLogger(__name__)
logger.setLevel(level=logging.WARNING)
Expand All @@ -41,7 +29,7 @@
"supported_https_auths": ["password", "token"],
"default_port": 8006,
"token_separator": "=",
"ssh_additional_options": ["--output-format", "json"],
"cli_additional_options": ["--output-format", "json"],
},
"PMG": {
"supported_backends": ["local", "https", "openssh", "ssh_paramiko"],
Expand All @@ -62,15 +50,31 @@ def config_failure(message, *args):


class ResourceException(Exception):
"""
An Exception thrown when an Proxmox API call failed
"""

def __init__(self, status_code, status_message, content, errors=None):
"""
Create a new ResourceException

:param status_code: The HTTP status code (faked by non-HTTP backends)
:type status_code: int
:param status_message: HTTP Status code (faked by non-HTTP backends)
:type status_message: str
:param content: Extended information on what went wrong
:type content: str
:param errors: Any specific errors that were encountered (converted to string), defaults to None
:type errors: Optional[object], optional
"""
self.status_code = status_code
self.status_message = status_message
self.content = content
self.errors = errors
if errors is not None:
content += " - {0}".format(errors)
message = "{0} {1}: {2}".format(status_code, status_message, content).strip()
super(ResourceException, self).__init__(message)
content += f" - {errors}"
message = f"{status_code} {status_message}: {content}".strip()
super().__init__(message)


class ProxmoxResource(object):
Expand All @@ -89,14 +93,14 @@ def __getattr__(self, item):
def url_join(self, base, *args):
scheme, netloc, path, query, fragment = urlparse.urlsplit(base)
path = path if len(path) else "/"
path = posixpath.join(path, *[("%s" % x) for x in args])
path = posixpath.join(path, *[str(x) for x in args])
return urlparse.urlunsplit([scheme, netloc, path, query, fragment])

def __call__(self, resource_id=None):
if resource_id in (None, ""):
return self

if isinstance(resource_id, basestring):
if isinstance(resource_id, (bytes, str)):
resource_id = resource_id.split("/")
elif not isinstance(resource_id, (tuple, list)):
resource_id = [str(resource_id)]
Expand All @@ -110,9 +114,9 @@ def __call__(self, resource_id=None):
def _request(self, method, data=None, params=None):
url = self._store["base_url"]
if data:
logger.info("%s %s %r", method, url, data)
logger.info(f"{method} {url} {data}")
else:
logger.info("%s %s", method, url)
logger.info(f"{method} {url}")

# passing None values to pvesh command breaks it, let's remove them just as requests library does
# helpful when dealing with function default values higher in the chain, no need to clean up in multiple places
Expand All @@ -129,7 +133,7 @@ def _request(self, method, data=None, params=None):
del data[key]

resp = self._store["session"].request(method, url, data=data, params=params)
logger.debug("Status code: %s, output: %s", resp.status_code, resp.content)
logger.debug(f"Status code: {resp.status_code}, output: {resp.content}")

if resp.status_code >= 400:
if hasattr(resp, "reason"):
Expand Down Expand Up @@ -173,7 +177,7 @@ def set(self, *args, **data):

class ProxmoxAPI(ProxmoxResource):
def __init__(self, host=None, backend="https", service="PVE", **kwargs):
super(ProxmoxAPI, self).__init__(**kwargs)
super().__init__(**kwargs)
service = service.upper()
backend = backend.lower()

Expand All @@ -194,7 +198,7 @@ def __init__(self, host=None, backend="https", service="PVE", **kwargs):
kwargs["service"] = service

# load backend module
self._backend = importlib.import_module(".backends.%s" % backend, "proxmoxer").Backend(
self._backend = importlib.import_module(f".backends.{backend}", "proxmoxer").Backend(
**kwargs
)
self._backend_name = backend
Expand Down