proxmox/plugins/module_utils/proxmox_datacenter_realm.py

from dataclasses import dataclass
from typing import List, Tuple

from ansible_collections.finallycoffee.proxmox.plugins.module_utils.common import _proxmox_request, ProxmoxAuthInfo


@dataclass(frozen=True)
class ProxmoxRealm:
    type: str
    realm: str
    comment: str = None
    default: bool = False


def get_realms(auth_info: ProxmoxAuthInfo) -> List['ProxmoxRealm']:
    realm_answer = _proxmox_request('get', '/access/domains', auth_info).json()
    return list(map(
        lambda r: ProxmoxRealm(r['type'], r['realm'], r.get('comment', None), r.get('default', False)),
        realm_answer['data']
    ))


def set_realm_data(auth_info: ProxmoxAuthInfo, realm: str, config: dict[str, str], dry_run: bool, state: str = 'present') -> Tuple[dict, dict]:
    existing_realm = _proxmox_request('get', f"/access/domains/{realm}", auth_info)
    existing_realm_data = None
    new_realm_data = {}
    if state == 'present':
        if existing_realm.ok:
            existing_realm_data = existing_realm.json()['data']
            if not dry_run:
                put_args_denylist = ['type']
                put_config = {k: v for k, v in config.items() if k not in put_args_denylist}
                realm_res = _proxmox_request('put', f"/access/domains/{realm}", auth_info, data=put_config)
                realm_res.raise_for_status()
        else:
            if not dry_run:
                realm_res = _proxmox_request('post', '/access/domains', auth_info, data=(config | {'realm': realm}))
                realm_res.raise_for_status()
        if not dry_run:
            if realm_res.ok:
                new_realm_data = _proxmox_request('get', f"/access/domains/{realm}", auth_info).json()['data']
        else:
            new_realm_data = config | {'name': realm}
    else:
        if existing_realm.ok:
            existing_realm_data = existing_realm.json()['data']
            if not dry_run:
                realm_res = _proxmox_request('delete', f"/access/domains/{realm}", auth_info)
                realm_res.raise_for_status()
                if realm_res.ok:
                    new_realm_data = None
            else:
                new_realm_data = None
        else:
            new_realm_data = None
    return existing_realm_data, new_realm_data


def do_realm_sync(auth_info: ProxmoxAuthInfo, realm: str, config: dict[str, str], dry_run: bool):
    sync_config = config | {"dry-run": dry_run}
    realm_sync = _proxmox_request('post', f"/access/domains/{realm}/sync", auth_info, data=sync_config)
    return realm_sync.ok