Skip to content

Feat/pbscontrol #77

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

Draft
lcaflc wants to merge 2 commits into
base: main
Choose a base branch
from
Draft

Feat/pbscontrol #77

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
51e430c
feat(pbscontrol): First version of pbscontrol
lcaflc Apr 7, 2026
8ab18fe
chore(pbscontrol): Add basic tests
lcaflc Apr 7, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
1 change: 1 addition & 0 deletions setup.cfg
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -25,3 +25,4 @@ where=src
[options.entry_points]
console_scripts =
pvecontrol = pvecontrol:main
pbscontrol = pbscontrol:main
157 changes: 157 additions & 0 deletions src/pbscontrol/__init__.py
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -0,0 +1,157 @@
#!/usr/bin/env python3

import sys
import logging
import signal

from types import SimpleNamespace
from importlib.metadata import version, PackageNotFoundError

import click
import urllib3

from pbscontrol import actions
from pvecontrol.utils import OutputFormats


def get_leaf_command(cmd, ctx, args):
if len(args) == 0:
return cmd, []

parser = cmd.make_parser(ctx)
_, args_without_options, _ = parser.parse_args(list(args))

if len(args_without_options) == 0:
return cmd, args

name, sub_cmd, sub_args = cmd.resolve_command(ctx, args_without_options)
if isinstance(sub_cmd, click.MultiCommand) and len(sub_args) > 0:
sub_ctx = sub_cmd.make_context(name, sub_args, parent=ctx)
return get_leaf_command(sub_cmd, sub_ctx, sub_args)

return sub_cmd, sub_args


class IgnoreRequiredForHelp(click.Group):
def __init__(self, *args, **kwargs):
super().__init__(*args, **kwargs)
self.ignoring = False

def _is_defaulting_to_help(self, ctx, args):
try:
leaf_cmd, leaf_args = get_leaf_command(self, ctx, args)

if leaf_cmd is None or leaf_cmd is self:
return False

return (
"--help" in leaf_args
or (isinstance(leaf_cmd, click.MultiCommand) and not leaf_cmd.invoke_without_command)
or (leaf_cmd.no_args_is_help and len(leaf_args) == 0)
)
except click.exceptions.UsageError:
return False

def parse_args(self, ctx, args):
if self._is_defaulting_to_help(ctx, args):
self.ignoring = True
for param in self.params:
param.required = False

return super().parse_args(ctx, args)

def format_commands(self, ctx, formatter) -> None:
commands = []
for subcommand in self.list_commands(ctx):
cmd = self.get_command(ctx, subcommand)
if cmd is None or cmd.hidden:
continue

commands.append((subcommand, cmd))

if len(commands) > 0:
limit = formatter.width - 6 - max(len(cmd[0]) for cmd in commands)

rows = []
for subcommand, cmd in commands:
if not isinstance(cmd, click.MultiCommand):
cmd_help = cmd.get_short_help_str(limit)
rows.append((subcommand, cmd_help))
continue
for subsubcommand in cmd.list_commands(ctx):
subcmd = cmd.get_command(ctx, subsubcommand)
if subcmd is None or subcmd.hidden:
continue
cmd_help = subcmd.get_short_help_str(limit)
rows.append((f"{subcommand} {subsubcommand}", cmd_help))

if rows:
with formatter.section("Commands"):
formatter.write_dl(rows)


try:
_version = version("pvecontrol")
except PackageNotFoundError:
_version = "unknown"


@click.group(
cls=IgnoreRequiredForHelp,
help=f"Proxmox Backup Server control CLI, version: {_version}",
epilog="Made with love by Enix.io",
)
@click.option("-d", "--debug", is_flag=True)
@click.option(
"-o",
"--output",
type=click.Choice([o.value for o in OutputFormats]),
show_default=True,
default=OutputFormats.TEXT.value,
callback=lambda *v: OutputFormats(v[2]),
)
@click.option(
"-s",
"--server",
metavar="NAME",
envvar="SERVER",
required=True,
help="Proxmox Backup Server name as defined in configuration",
)
@click.option(
"--unicode/--no-unicode",
envvar="UNICODE",
default=True,
help="Use unicode characters for output",
)
@click.option(
"--color/--no-color",
envvar="COLOR",
default=True,
help="Use colorized output",
)
@click.pass_context
def pbscontrol(ctx, debug, output, server, unicode, color):
signal.signal(signal.SIGINT, lambda *_: sys.exit(130))
urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)

if not ctx.command.ignoring:
args = SimpleNamespace(output=output, server=server, unicode=unicode, color=color)

logging.basicConfig(encoding="utf-8", level=logging.DEBUG if debug else logging.INFO)
logging.debug("Arguments: %s", args)

ctx.ensure_object(dict)
ctx.obj["args"] = args


pbscontrol.add_command(cmd=actions.server.status, name="status")


def main():
# pylint: disable=no-value-for-parameter
pbscontrol(auto_envvar_prefix="PBSCONTROL")


if __name__ == "__main__":
sys.exit(main())
4 changes: 4 additions & 0 deletions src/pbscontrol/__main__.py
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -0,0 +1,4 @@
import pbscontrol

if __name__ == "__main__":
pbscontrol.main()
1 change: 1 addition & 0 deletions src/pbscontrol/actions/__init__.py
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -0,0 +1 @@
from pbscontrol.actions import server
40 changes: 40 additions & 0 deletions src/pbscontrol/actions/server.py
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -0,0 +1,40 @@
import click

from humanize import naturalsize

from pbscontrol.models.server import PBSServer
from pvecontrol.utils import OutputFormats, render_output


@click.command()
@click.pass_context
def status(ctx):
"""Show Proxmox Backup Server status"""
pbs = PBSServer.create_from_config(ctx.obj["args"].server)
usage = pbs.datastore_usage

if ctx.obj["args"].output == OutputFormats.TEXT:
print(f"""\n\
Name: {pbs.name}
Version: {pbs.version.get('version', 'unknown')}
Datastores:""")
for ds in usage:
total = naturalsize(ds["total"], binary=True, format="%.2f")
used = naturalsize(ds["used"], binary=True, format="%.2f")
avail = naturalsize(ds["avail"], binary=True, format="%.2f")
percent = ds["used"] / ds["total"] * 100 if ds["total"] else 0
print(f" {ds['store']}: {used}/{total} ({percent:.1f}%), available: {avail}, gc: {ds['gc-status']}")
print()
else:
render_table = [
{
"store": ds["store"],
"total": ds["total"],
"used": ds["used"],
"avail": ds["avail"],
"percent": round(ds["used"] / ds["total"] * 100 if ds["total"] else 0, 1),
"gc-status": ds["gc-status"],
}
for ds in usage
]
print(render_output(render_table, output=ctx.obj["args"].output))
50 changes: 50 additions & 0 deletions src/pbscontrol/config.py
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -0,0 +1,50 @@
import logging
import sys
import confuse

configtemplate = {
"servers": confuse.Sequence( # pylint: disable=abstract-class-instantiated
{
"name": str,
"host": str,
"user": str,
"password": confuse.Optional(str, None),
"token_name": confuse.Optional(str, None),
"token_value": confuse.Optional(str, None),
"proxy_certificate": confuse.Optional(
confuse.OneOf(
[
str,
{
"cert": str,
"key": str,
},
]
),
None,
),
"port": confuse.Optional(int, default=8007),
"timeout": confuse.Optional(int, default=60),
"ssl_verify": confuse.Optional(bool, False),
}
),
}


config = confuse.LazyConfig("pbscontrol", __name__)


def set_config(server_name):
validconfig = config.get(configtemplate)
logging.debug("configuration is %s", validconfig)

serverconfig = False
for s in validconfig.servers:
if s.name == server_name:
serverconfig = s
if not serverconfig:
logging.error('No such server "%s"', server_name)
sys.exit(1)
logging.debug("serverconfig is %s", serverconfig)

return serverconfig
Empty file added src/pbscontrol/models/__init__.py
View file
Open in desktop
Empty file.
61 changes: 61 additions & 0 deletions src/pbscontrol/models/server.py
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -0,0 +1,61 @@
import logging
import sys

from proxmoxer import ProxmoxAPI
from requests.exceptions import SSLError

from pvecontrol.utils import run_auth_commands
from pbscontrol.config import set_config


class PBSServer:
"""Proxmox Backup Server"""

def __init__(self, name, host, port, timeout, verify_ssl=False, **auth):
try:
self.api = ProxmoxAPI(host, port=port, timeout=timeout, verify_ssl=verify_ssl, service="pbs", **auth)
except SSLError as e:
print(e)
sys.exit(1)
self.name = name
self._initstatus()

def _initstatus(self):
self.version = self.api.version.get()
self.datastores = self.api.admin.datastore.get()

@staticmethod
def create_from_config(server_name):
logging.info("Proxmox Backup Server: %s", server_name)

serverconfig = set_config(server_name)
auth = run_auth_commands(serverconfig)
return PBSServer(
serverconfig.name,
serverconfig.host,
port=serverconfig.port,
verify_ssl=serverconfig.ssl_verify,
timeout=serverconfig.timeout,
**auth,
)

@property
def datastore_usage(self):
usage = []
for ds in self.datastores:
store = ds.get("store")
try:
status = self.api.admin.datastore(store).status.get()
except Exception as e: # pylint: disable=broad-exception-caught
logging.warning("Could not get status for datastore %s: %s", store, e)
status = {}
usage.append(
{
"store": store,
"total": status.get("total", 0),
"used": status.get("used", 0),
"avail": status.get("avail", 0),
"gc-status": status.get("gc-status", {}).get("error", "ok"),
}
)
return usage
Empty file added src/tests/pbscontrol/__init__.py
View file
Open in desktop
Empty file.
Empty file added src/tests/pbscontrol/fixtures/__init__.py
View file
Open in desktop
Empty file.
Loading
Loading