1#@!#!123s

: / usr / lib / rads / venv / lib64 / python3.13 / site-packages / cpapis /
Filename : _uapi.py
back
"""uapi functions"""
from typing import Union
from ._base import CpAPIBase
from ._errors import CpAPIErrorMsg


class Uapi(CpAPIBase):
    """uapi functions"""

    __module__ = 'cpapis'

    def __init__(self):
        if self._can_exec('/usr/bin/uapi'):
            super().__init__('/usr/bin/uapi')
        else:
            super().__init__('/usr/local/cpanel/bin/uapi')

    def __call__(
        self,
        module: str,
        user: Union[str, None] = None,
        args: Union[dict, None] = None,
        timeout: Union[float, None] = None,
        *,
        check: bool = False,
    ):
        """Query cPanel uapi

        Args:
            module: uapi module to use in format x::y
            user: ``--user`` arg to use
            args: key-vals to send to uapi
            timeout: timeout for the uapi command in secs
            check: check .result.status from the result and raise
                an CpAPIErrorMsg on failure. Defaults False.

        Raises:
            CpAPIExecFail: command failed to execute or returned invalid json
            CpAPIErrorMsg: check=True was set and the API reported an error
        """
        ret = self._exec(
            module_args=module.split('::'),
            user=user,
            args=args,
            timeout=timeout,
        )
        if check:
            self.check(ret)
        return ret

    @classmethod
    def check(cls, data: dict) -> None:
        """Parse output common to most uapi functions and raise an
        exception if the function failed

        Args:
            data: result data from a uapi call

        Raises:
            CpAPIErrorMsg: the API reported an error
        """
        try:
            if data['result']['status'] != 1:
                msg = data['result'].get("errors", None)
                if not msg:
                    msg = data['result'].get('messages', None)
                if not msg:
                    msg = "unspecified error"
                if isinstance(msg, list):
                    msg = '. '.join(map(str, msg))
                else:
                    msg = str(msg)
                raise CpAPIErrorMsg(msg=msg, data=data)
        except (TypeError, KeyError) as exc:
            raise CpAPIErrorMsg(
                msg=f"{type(exc).__name__}: {exc}", data=data
            ) from exc

    def create_mysql_db(
        self, user: str, dbname: str, timeout: Union[float, None] = None
    ) -> None:
        """Create a MySQL database

        Args:
            user: cPanel user to assign the database to
            dbname: database name including its prefix
            timeout: timeout for the uapi command in secs

        Raises:
            CpAPIExecFail: command failed to execute or returned invalid json
            CpAPIErrorMsg: the API reported an error
        """
        self(
            module='Mysql::create_database',
            user=user,
            args={'name': dbname},
            check=True,
            timeout=timeout,
        )

    def create_pgsql_db(
        self, user: str, dbname: str, timeout: Union[float, None] = None
    ) -> None:
        """Create a PostgreSQL database

        Args:
            user: cPanel user to assign the database to
            dbname: database name including its prefix
            timeout: timeout for the uapi command in secs

        Raises:
            CpAPIExecFail: command failed to execute or returned invalid json
            CpAPIErrorMsg: the API reported an error
        """
        self(
            module='Postgresql::create_database',
            user=user,
            args={'name': dbname},
            check=True,
            timeout=timeout,
        )