refactor(proxmox): unify ProxmoxClient construction via from_server
Some checks failed
Backend Tests / pytest (Python 3.12) (push) Failing after 48s
Deploy to Production / Trigger production update (push) Successful in 8s

Collapse four duplicated host+auth client-building blocks into a single
ProxmoxClient.from_server() classmethod (the canonical source of truth):

- _helpers._get_proxmox_client now delegates to from_server (107 callers
  untouched)
- ipam._build_proxmox_client removed; its one call site uses from_server
- task_queue_service keeps its per-server cache pool, builds via from_server
- monitoring_worker keeps its ValueError-on-missing-auth contract, builds
  via from_server

Behaviour preserved (token-first, then password; non-default port appended).
Verified: containers rebuild healthy, smoke test of both auth paths passes.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
2026-06-19 22:30:57 +05:00
parent 34fe9b4924
commit d66928ee78
5 changed files with 44 additions and 77 deletions

View File

@@ -83,6 +83,34 @@ def get_cached_client(host: str, user: str, password_hash: str = None,
)
class ProxmoxClient(VmMixin, LxcMixin, ClusterMixin, StorageMixin, NetworkMixin, SnapshotMixin):
@classmethod
def from_server(cls, server) -> "ProxmoxClient":
"""Single source of truth for building a client from a ProxmoxServer model.
Picks token auth when available, otherwise password auth, and appends a
non-default port to the host. Callers across the API and services layers
(see api/proxmox/_helpers.py and services/task_queue_service.py) should
go through here instead of duplicating the host/auth branching.
"""
host = server.ip_address or getattr(server, "hostname", "") or ""
if getattr(server, "port", None) and server.port != 8006:
host = f"{host}:{server.port}"
if getattr(server, "use_password", False) and getattr(server, "password", ""):
return cls(
host=host,
user=server.api_user,
password=server.password,
verify_ssl=server.verify_ssl,
)
return cls(
host=host,
user=server.api_user,
token_name=server.api_token_name,
token_value=server.api_token_value,
verify_ssl=server.verify_ssl,
)
def __init__(self, host: str, user: str = "root@pam", password: str = None,
token_name: str = None, token_value: str = None, verify_ssl: bool = True,
timeout: int = 30):