cloudfloordns.client.sync.domain

  1# from dataclasses import dataclass, field
  2from collections import ChainMap
  3from typing import Optional, Union
  4
  5from cloudfloordns.models import Contact, Domain, DomainPayload, TLDPrice
  6
  7DEFAULT_PRIMARY_NS = "ns1.g02.cfdns.net"
  8DEFAULT_LIMIT = 99999
  9
 10
 11class Domains:
 12    def __init__(self, client) -> None:
 13        self._client = client
 14
 15    def _raw_pricelist(self, timeout=None) -> list:
 16        """
 17        List all domains
 18        """
 19        url = "/domain/tld/pricelist"
 20        return self._client.get(
 21            url,
 22            # data=data,
 23            timeout=timeout,
 24        )
 25
 26    def pricelist(self, timeout=None) -> list[TLDPrice]:
 27        """
 28        List all domains
 29        """
 30        res = self._raw_pricelist(timeout=timeout)
 31        return [TLDPrice.model_validate(p) for p in res]
 32
 33    def get_by_name(self, domainname, zone_enabled=None, timeout=None):
 34        res = self.list(
 35            zone_enabled=zone_enabled,
 36            timeout=timeout,
 37        )
 38        return next((r for r in res if r.domainname == domainname), None)
 39
 40    def delete(self, domain: str, timeout=None):
 41        url = f"/domain/{domain}/delete_domain"
 42        return self._client.delete(
 43            url,
 44            timeout=timeout,
 45        )
 46
 47    def raw_get(self, domain, timeout=None) -> Domain:
 48        url = f"/domain/{domain}"
 49        return self._client.get(
 50            url,
 51            timeout=timeout,
 52        )
 53
 54    def get(self, domain, timeout=None) -> Domain:
 55        res = self.raw_get(domain, timeout=timeout)
 56        return Domain.model_validate(res)
 57
 58    def raw_register(self, payload: DomainPayload, timeout=None) -> dict:
 59        """
 60        This will register a domain on CloudfloorDNS.
 61        The registration can be cancelled during a short period (to be confirmed)
 62        """
 63        url = "/domain"
 64        return self._client.post(
 65            url,
 66            data=payload.model_dump(by_alias=True),
 67            timeout=timeout,
 68        )
 69
 70    def raw_registry_info(self, domain, timeout=None) -> dict:
 71        url = f"/domain/{domain}/get_registry_info"
 72        return self._client.get(
 73            url,
 74            timeout=timeout,
 75        )
 76
 77    def registry_info(self, domain, timeout=None) -> Domain:
 78        """
 79        Return the data currently stored in the registry.
 80        NOTE:
 81        - The result may differ from `get` method (e.g. if the data push failed)
 82        - This is not similar as a WHOIS command since it retrieves redacted data as well
 83        """
 84        res = self.raw_registry_info(domain, timeout=timeout)
 85        return Domain.model_validate(res)
 86
 87    def raw_status(self, domain, timeout=None) -> dict:
 88        url = f"/domain/{domain}/status"
 89        return self._client.get(
 90            url,
 91            timeout=timeout,
 92        )
 93
 94    def raw_update(
 95        self, domain: str, payload: Union[Domain, DomainPayload], timeout=None
 96    ):
 97        if isinstance(payload, Domain):
 98            payload = payload.register_payload()
 99        # print(f"Trying to update domain {domain} with payload:\n{json.dumps(payload, indent=4)}")
100        url = f"/domain/{domain}"
101        data = payload.dump_for_update()
102        return self._client.patch(
103            url,
104            data=data,
105            timeout=timeout,
106        )
107
108    def _raw_list(self, limit=None, offset=None, timeout=None) -> list:
109        """
110        List all domains
111        """
112        data = {"limit": limit or DEFAULT_LIMIT}
113        if offset:
114            data["offset"] = offset
115        url = "/domain"
116        return self._client.get(
117            url,
118            data=data,
119            timeout=timeout,
120        )
121
122    def list(self, limit=None, offset=None, timeout=None) -> list[Domain]:
123        """
124        List all domains
125        """
126        res = self._raw_list(limit=limit, offset=offset, timeout=timeout)
127        return [Domain.model_validate(d) for d in res]
128
129    # Only updating contacts does not seem to work
130    def update_contact(
131        self,
132        domain: str,
133        owner: Optional[Contact] = None,
134        admin: Optional[Contact] = None,
135        tech: Optional[Contact] = None,
136        bill: Optional[Contact] = None,
137        timeout=None,
138    ):
139        """
140        This method works according to the documentation, but some required parameters might be missing.
141        """
142        converted = (
143            d
144            for d in (
145                owner and owner.as_owner(),
146                admin and admin.as_admin(),
147                tech and tech.as_tech(),
148                bill and bill.as_bill(),
149            )
150            if d
151        )
152        payload = dict(ChainMap(*converted))
153        # print(f"Trying to update domain {domain} with payload:\n{json.dumps(payload, indent=4)}")
154        url = f"/domain/{domain}"
155        return self._client.patch(
156            url,
157            data=payload,
158            timeout=timeout,
159        )
160
161
162__all__ = [
163    "Domains",
164]