ispconfig-py/src/ispconfig/client.py

"""``ISPConfigClient`` — top-level entry point, session lifecycle, submodule wiring."""

from __future__ import annotations

import logging
from types import TracebackType
from typing import Any

from . import exceptions as _exc
from ._soap import SoapFault, SoapTransport
from .admin import AdminModule
from .aps import ApsModule
from .backups import BackupsModule
from .clients import ClientsModule
from .cron import CronModule
from .databases import DatabasesModule
from .dns import DnsModule
from .domains import DomainsModule
from .ftp import FtpModule
from .mail import MailModule
from .misc import MiscModule
from .monitor import MonitorModule
from .openvz import OpenvzModule
from .server import ServerModule
from .shell import ShellModule
from .sites import SitesModule
from .webdav import WebdavModule

log = logging.getLogger("ispconfig")


class ISPConfigClient:
    """High-level client for the ISPConfig remote SOAP API.

    Use as a context manager to auto-login on enter and auto-logout on exit::

        with ISPConfigClient(url, "admin", "password") as c:
            site = c.sites.web_domain_get(156)
            print(site["domain"])

    Session IDs are managed internally and never returned to callers.
    If a call fails with a session-expired fault, the client re-authenticates
    once and retries (controlled by ``max_retries``).
    """

    def __init__(
        self,
        url: str,
        username: str,
        password: str,
        *,
        verify_ssl: bool = True,
        timeout: float = 30.0,
        max_retries: int = 1,
    ) -> None:
        self._url = url
        self._username = username
        self._password = password
        self._max_retries = max_retries
        self._transport = SoapTransport(url, verify_ssl=verify_ssl, timeout=timeout)
        self._session_id: str | None = None

        # Hand-audited modules (stable API).
        self.sites = SitesModule(self)
        self.dns = DnsModule(self)
        self.mail = MailModule(self)
        self.databases = DatabasesModule(self)
        self.clients = ClientsModule(self)
        # Auto-generated modules (full surface, param shapes not yet
        # verified in prod use — see per-method docstrings).
        self.admin = AdminModule(self)
        self.aps = ApsModule(self)
        self.backups = BackupsModule(self)
        self.cron = CronModule(self)
        self.domains = DomainsModule(self)
        self.ftp = FtpModule(self)
        self.misc = MiscModule(self)
        self.monitor = MonitorModule(self)
        self.openvz = OpenvzModule(self)
        self.server = ServerModule(self)
        self.shell = ShellModule(self)
        self.webdav = WebdavModule(self)

    # ---- context manager ---------------------------------------------

    def __enter__(self) -> ISPConfigClient:
        self.login()
        return self

    def __exit__(
        self,
        exc_type: type[BaseException] | None,
        exc: BaseException | None,
        tb: TracebackType | None,
    ) -> None:
        try:
            self.logout()
        except Exception as e:  # pragma: no cover — cleanup path, never raise out
            log.warning("logout on __exit__ failed: %s", e)

    # ---- session lifecycle -------------------------------------------

    def login(self) -> None:
        """Open a session. Stores the session id internally."""
        try:
            sid = self._transport.call(
                "login",
                (("username", self._username), ("password", self._password)),
            )
        except SoapFault as f:
            raise _exc.map_fault(f.faultcode, f.faultstring) from f
        if not isinstance(sid, str) or not sid:
            raise _exc.AuthError("login returned empty session id")
        self._session_id = sid
        log.debug("ispconfig login ok (session %s...)", sid[:8])

    def logout(self) -> bool:
        """Close the session. Safe to call even if never logged in."""
        if self._session_id is None:
            return False
        sid = self._session_id
        self._session_id = None
        try:
            result = self._transport.call("logout", (("session_id", sid),))
        except SoapFault as f:
            log.debug("logout fault ignored: %s", f)
            return False
        return bool(result)

    @property
    def session_id(self) -> str | None:
        """Read-only accessor — exposed for debugging, not for API calls."""
        return self._session_id

    # ---- escape hatches ----------------------------------------------

    def raw_call(self, method: str, *args: Any) -> Any:
        """Invoke an arbitrary ISPConfig remote method by name.

        Use this when the SDK doesn't yet wrap the method you need —
        newer ISPConfig versions may expose calls our inventory hasn't
        caught up with. Args are passed positionally; names are cosmetic
        on the wire, so we auto-number them as ``arg1``, ``arg2``, ...

        If the call fails, capture ``FaultError.faultcode`` /
        ``FaultError.faultstring`` and file an issue against
        ``Sulkta-OSS/ispconfig-py`` so we can add the method properly.
        """
        named_args = tuple((f"arg{i + 1}", v) for i, v in enumerate(args))
        return self._call(method, *named_args)

    def list_functions(self) -> list[str]:
        """Introspect the panel: return the list of remote methods it exposes.

        Wrapper for ISPConfig's own ``get_function_list``. Handy when
        checking whether your panel version supports a given call before
        attempting it via :meth:`raw_call`.
        """
        result = self._call("get_function_list")
        if isinstance(result, list):
            return [str(x) for x in result]
        if isinstance(result, dict):
            # Some versions return a map keyed by integer index.
            return [str(v) for v in result.values()]
        return []

    # ---- the hot path ------------------------------------------------

    def _call(self, method: str, *args: tuple[str, Any]) -> Any:
        """Invoke ``method(session_id, *args)`` with typed error mapping + retry.

        This is what the submodules call. It prepends the session id,
        translates SOAP faults, and retries once on session expiry.
        """
        attempts = 0
        while True:
            if self._session_id is None:
                self.login()
            sid_arg = ("session_id", self._session_id or "")
            try:
                return self._transport.call(method, (sid_arg, *args))
            except SoapFault as f:
                mapped = _exc.map_fault(f.faultcode, f.faultstring)
                if _exc.is_session_expired(mapped) and attempts < self._max_retries:
                    log.info("ispconfig session expired, re-authenticating")
                    self._session_id = None
                    attempts += 1
                    continue
                raise mapped from f