Skip to content

OctoDNS Metaname Module

Reference documentation generated directly from the module docstrings.

Client

Thin wrapper around the Metaname JSON-RPC API used by OctoDNS.

Contact dataclass

Contact details used when provisioning domains via the API.

Source code in octodns_metaname/client.py 
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
@dataclass
class Contact:
    """Contact details used when provisioning domains via the API."""

    name: str
    email: str
    phone_country_code: str
    phone_area_code: Optional[str]
    phone_local_number: str
    organisation: Optional[str] = None
    address_line1: str = "123 Test Street"
    address_line2: Optional[str] = None
    city: str = "Wellington"
    region: Optional[str] = None
    postal_code: str = "6011"
    country_code: str = "NZ"

    def to_payload(self) -> Dict[str, Any]:
        """Serialise the contact into the structure expected by Metaname."""

        return {
            "name": self.name,
            "email_address": self.email,
            "organisation_name": self.organisation,
            "postal_address": {
                "line1": self.address_line1,
                "line2": self.address_line2,
                "city": self.city,
                "region": self.region,
                "postal_code": self.postal_code,
                "country_code": self.country_code,
            },
            "phone_number": {
                "country_code": self.phone_country_code,
                "area_code": self.phone_area_code,
                "local_number": self.phone_local_number,
            },
            "fax_number": None,
        }

to_payload()

Serialise the contact into the structure expected by Metaname.

Source code in octodns_metaname/client.py 
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
def to_payload(self) -> Dict[str, Any]:
    """Serialise the contact into the structure expected by Metaname."""

    return {
        "name": self.name,
        "email_address": self.email,
        "organisation_name": self.organisation,
        "postal_address": {
            "line1": self.address_line1,
            "line2": self.address_line2,
            "city": self.city,
            "region": self.region,
            "postal_code": self.postal_code,
            "country_code": self.country_code,
        },
        "phone_number": {
            "country_code": self.phone_country_code,
            "area_code": self.phone_area_code,
            "local_number": self.phone_local_number,
        },
        "fax_number": None,
    }

ZoneRecord dataclass

Representation of a DNS record as returned by the Metaname API.

Source code in octodns_metaname/client.py 
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
@dataclass
class ZoneRecord:
    """Representation of a DNS record as returned by the Metaname API."""

    reference: Optional[str]
    name: str
    rtype: str
    data: str
    ttl: int
    aux: Optional[int] = None

    @classmethod
    def from_api(cls, payload: Dict[str, Any]) -> "ZoneRecord":
        """Construct a zone record from an API payload."""

        return cls(
            reference=payload.get("reference"),
            name=payload.get("name") or "@",
            rtype=payload["type"].upper(),
            data=payload.get("data", ""),
            ttl=int(payload.get("ttl", 3600)),
            aux=payload.get("aux"),
        )

    def to_api_payload(self) -> Dict[str, Any]:
        """Serialise the record into the JSON-RPC payload schema."""

        payload: Dict[str, Any] = {
            "name": self.name,
            "type": self.rtype,
            "data": self.data,
            "ttl": self.ttl,
        }
        if self.aux is not None:
            payload["aux"] = self.aux
        return payload

from_api(payload) classmethod

Construct a zone record from an API payload.

Source code in octodns_metaname/client.py 
68
69
70
71
72
73
74
75
76
77
78
79
@classmethod
def from_api(cls, payload: Dict[str, Any]) -> "ZoneRecord":
    """Construct a zone record from an API payload."""

    return cls(
        reference=payload.get("reference"),
        name=payload.get("name") or "@",
        rtype=payload["type"].upper(),
        data=payload.get("data", ""),
        ttl=int(payload.get("ttl", 3600)),
        aux=payload.get("aux"),
    )

to_api_payload()

Serialise the record into the JSON-RPC payload schema.

Source code in octodns_metaname/client.py 
81
82
83
84
85
86
87
88
89
90
91
92
def to_api_payload(self) -> Dict[str, Any]:
    """Serialise the record into the JSON-RPC payload schema."""

    payload: Dict[str, Any] = {
        "name": self.name,
        "type": self.rtype,
        "data": self.data,
        "ttl": self.ttl,
    }
    if self.aux is not None:
        payload["aux"] = self.aux
    return payload

MetanameError

Bases: RuntimeError

Generic error for Metaname client failures.

Source code in octodns_metaname/client.py 
95
96
class MetanameError(RuntimeError):
    """Generic error for Metaname client failures."""

MetanameAPIError

Bases: MetanameError

Raised when the remote API reports an error.

Source code in octodns_metaname/client.py 
 99
100
101
102
103
104
105
106
107
class MetanameAPIError(MetanameError):
    """Raised when the remote API reports an error."""

    def __init__(self, message: str, *, code: Optional[int] = None, payload: Any = None) -> None:
        self.code = code
        self.payload = payload
        if code is not None:
            message = f"{message} (code {code})"
        super().__init__(message)

MetanameClient

Convenience wrapper around Metaname's JSON-RPC 2.0 endpoints.

Source code in octodns_metaname/client.py 
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
class MetanameClient:
    """Convenience wrapper around Metaname's JSON-RPC 2.0 endpoints."""

    def __init__(self, *, base_url: str = TEST_API_URL, timeout: float = 10.0) -> None:
        """
        Parameters
        ----------
        base_url:
            Target API URL. Defaults to the Metaname test endpoint.
        timeout:
            Timeout (seconds) applied to HTTP requests.
        """

        self.base_url = base_url.rstrip("/")
        self.timeout = timeout
        self.account_ref = get_secret("METANAME_ACCOUNT_REF")
        self.api_key = get_secret("METANAME_API_TOKEN")

    def _rpc(self, method: str, params: list[Any], *, request_id: int = 1) -> Any:
        """Call a JSON-RPC method and return the parsed ``result`` payload."""

        payload = {
            "jsonrpc": "2.0",
            "method": method,
            "params": [self.account_ref, self.api_key, *params],
            "id": request_id,
        }
        try:
            response = requests.post(self.base_url, json=payload, timeout=self.timeout)
        except requests.RequestException as exc:  # pragma: no cover
            raise MetanameError(f"Request to Metaname failed: {exc}") from exc
        if response.status_code != 200:
            raise MetanameError(f"Metaname returned HTTP {response.status_code}: {response.text}")
        try:
            data = response.json()
        except json.JSONDecodeError as exc:
            raise MetanameError("Metaname response was not valid JSON") from exc
        error = data.get("error")
        if error:
            raise MetanameAPIError(
                error.get("message", "Metaname API error"),
                code=error.get("code"),
                payload=error.get("data"),
            )
        if "result" not in data:
            raise MetanameError("Metaname API response missing 'result'")
        result = data["result"]
        if result is None:
            return {}
        if not isinstance(result, dict) and not isinstance(result, list):
            return {"value": result}
        return result

    def ping(self) -> Dict[str, Any]:
        """Check authentication by querying the account balance."""

        return self._rpc("account_balance", [])

    def list_zone_records(
        self, domain: str, *, page_size: Optional[int] = None
    ) -> list[ZoneRecord]:
        """
        Retrieve all DNS records for ``domain``.

        Parameters
        ----------
        domain:
            Fully-qualified domain (may end with a trailing dot).
        page_size:
            When provided, fetch records in chunks using ``dns_zone_chunk``.
        """

        return list(self.iter_zone_records(domain, page_size=page_size))

    def iter_zone_records(
        self, domain: str, *, page_size: Optional[int] = None
    ) -> Iterator[ZoneRecord]:
        """Yield DNS records for ``domain`` optionally using pagination."""

        domain = _strip_trailing_dot(domain)
        if page_size:
            offset = 0
            while True:
                records = self._rpc("dns_zone_chunk", [domain, page_size, offset])
                if not records:
                    break
                for item in records:
                    yield ZoneRecord.from_api(item)
                offset += len(records)
                if len(records) < page_size:
                    break
            return

        records = self._rpc("dns_zone", [domain])
        if isinstance(records, dict):
            records = records.get("records", [])
        for item in records or []:
            yield ZoneRecord.from_api(item)

    def create_zone_record(
        self, domain: str, record: ZoneRecord | Dict[str, Any]
    ) -> Dict[str, Any]:
        """Create a DNS record within ``domain``."""

        domain = _strip_trailing_dot(domain)
        payload = record.to_api_payload() if isinstance(record, ZoneRecord) else dict(record)
        return self._rpc("create_dns_record", [domain, payload])

    def update_zone_record(
        self,
        domain: str,
        reference: str,
        record: ZoneRecord | Dict[str, Any],
    ) -> Dict[str, Any]:
        """Update an existing record identified by ``reference``."""

        domain = _strip_trailing_dot(domain)
        payload = record.to_api_payload() if isinstance(record, ZoneRecord) else dict(record)
        return self._rpc("update_dns_record", [domain, reference, payload])

    def delete_zone_record(self, domain: str, reference: str) -> Dict[str, Any]:
        """Delete a record from ``domain`` by ``reference``."""

        domain = _strip_trailing_dot(domain)
        return self._rpc("delete_dns_record", [domain, reference])

    @staticmethod
    def _default_contact() -> Contact:
        try:
            email = get_secret("METANAME_CONTACT_EMAIL")
        except MissingSecret:
            email = os.getenv("METANAME_CONTACT_EMAIL", "team@startmeup.nz")

        name = _get_env_or_secret("METANAME_CONTACT_NAME", default="StartMeUp DNS")
        org = _get_env_or_secret("METANAME_CONTACT_ORG")
        phone_country = _get_env_or_secret("METANAME_CONTACT_PHONE_COUNTRY", default="64")
        phone_area = _get_env_or_secret("METANAME_CONTACT_PHONE_AREA")
        phone_local = _get_env_or_secret("METANAME_CONTACT_PHONE_LOCAL", default="2345678")

        return Contact(
            name=name,
            email=email,
            organisation=org,
            phone_country_code=phone_country,
            phone_area_code=phone_area,
            phone_local_number=phone_local,
        )

__init__(*, base_url=TEST_API_URL, timeout=10.0)

Parameters

base_url: Target API URL. Defaults to the Metaname test endpoint. timeout: Timeout (seconds) applied to HTTP requests.

Source code in octodns_metaname/client.py 
113
114
115
116
117
118
119
120
121
122
123
124
125
126
def __init__(self, *, base_url: str = TEST_API_URL, timeout: float = 10.0) -> None:
    """
    Parameters
    ----------
    base_url:
        Target API URL. Defaults to the Metaname test endpoint.
    timeout:
        Timeout (seconds) applied to HTTP requests.
    """

    self.base_url = base_url.rstrip("/")
    self.timeout = timeout
    self.account_ref = get_secret("METANAME_ACCOUNT_REF")
    self.api_key = get_secret("METANAME_API_TOKEN")

ping()

Check authentication by querying the account balance.

Source code in octodns_metaname/client.py 
163
164
165
166
def ping(self) -> Dict[str, Any]:
    """Check authentication by querying the account balance."""

    return self._rpc("account_balance", [])

list_zone_records(domain, *, page_size=None)

Retrieve all DNS records for domain.

Parameters

domain: Fully-qualified domain (may end with a trailing dot). page_size: When provided, fetch records in chunks using dns_zone_chunk.

Source code in octodns_metaname/client.py 
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
def list_zone_records(
    self, domain: str, *, page_size: Optional[int] = None
) -> list[ZoneRecord]:
    """
    Retrieve all DNS records for ``domain``.

    Parameters
    ----------
    domain:
        Fully-qualified domain (may end with a trailing dot).
    page_size:
        When provided, fetch records in chunks using ``dns_zone_chunk``.
    """

    return list(self.iter_zone_records(domain, page_size=page_size))

iter_zone_records(domain, *, page_size=None)

Yield DNS records for domain optionally using pagination.

Source code in octodns_metaname/client.py 
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
def iter_zone_records(
    self, domain: str, *, page_size: Optional[int] = None
) -> Iterator[ZoneRecord]:
    """Yield DNS records for ``domain`` optionally using pagination."""

    domain = _strip_trailing_dot(domain)
    if page_size:
        offset = 0
        while True:
            records = self._rpc("dns_zone_chunk", [domain, page_size, offset])
            if not records:
                break
            for item in records:
                yield ZoneRecord.from_api(item)
            offset += len(records)
            if len(records) < page_size:
                break
        return

    records = self._rpc("dns_zone", [domain])
    if isinstance(records, dict):
        records = records.get("records", [])
    for item in records or []:
        yield ZoneRecord.from_api(item)

create_zone_record(domain, record)

Create a DNS record within domain.

Source code in octodns_metaname/client.py 
209
210
211
212
213
214
215
216
def create_zone_record(
    self, domain: str, record: ZoneRecord | Dict[str, Any]
) -> Dict[str, Any]:
    """Create a DNS record within ``domain``."""

    domain = _strip_trailing_dot(domain)
    payload = record.to_api_payload() if isinstance(record, ZoneRecord) else dict(record)
    return self._rpc("create_dns_record", [domain, payload])

update_zone_record(domain, reference, record)

Update an existing record identified by reference.

Source code in octodns_metaname/client.py 
218
219
220
221
222
223
224
225
226
227
228
def update_zone_record(
    self,
    domain: str,
    reference: str,
    record: ZoneRecord | Dict[str, Any],
) -> Dict[str, Any]:
    """Update an existing record identified by ``reference``."""

    domain = _strip_trailing_dot(domain)
    payload = record.to_api_payload() if isinstance(record, ZoneRecord) else dict(record)
    return self._rpc("update_dns_record", [domain, reference, payload])

delete_zone_record(domain, reference)

Delete a record from domain by reference.

Source code in octodns_metaname/client.py 
230
231
232
233
234
def delete_zone_record(self, domain: str, reference: str) -> Dict[str, Any]:
    """Delete a record from ``domain`` by ``reference``."""

    domain = _strip_trailing_dot(domain)
    return self._rpc("delete_dns_record", [domain, reference])

Secrets Resolver

Secret resolution helpers for the Metaname provider.

By default secrets are pulled straight from environment variables. Users can optionally register a resolver (e.g., 1Password, Vault) via set_secret_resolver or the OCTODNS_METANAME_SECRET_RESOLVER env var, which should contain module:function. The resolver receives the secret name and the value of <NAME>_REF (if present) and should return the resolved secret or None when it cannot help.

MissingSecret

Bases: RuntimeError

Raised when a required secret cannot be resolved.

Source code in octodns_metaname/secrets.py 
21
22
class MissingSecret(RuntimeError):
    """Raised when a required secret cannot be resolved."""

set_secret_resolver(resolver)

Register a custom secret resolver (or clear when None).

Source code in octodns_metaname/secrets.py 
25
26
27
28
29
30
def set_secret_resolver(resolver: Optional[Resolver]) -> None:
    """Register a custom secret resolver (or clear when ``None``)."""

    global _secret_resolver, _resolver_loaded
    _secret_resolver = resolver
    _resolver_loaded = True

get_secret(name)

Resolve a secret via env var or configured resolver.

Source code in octodns_metaname/secrets.py 
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
def get_secret(name: str) -> str:
    """Resolve a secret via env var or configured resolver."""

    direct = os.getenv(name)
    if direct:
        return direct

    ref_env = f"{name}_REF"
    reference = os.getenv(ref_env)

    _ensure_resolver_loaded()

    if _secret_resolver:
        resolved = _secret_resolver(name, reference)
        if resolved:
            return resolved

    if reference:
        raise MissingSecret(
            f"Secret reference provided via {ref_env} but no resolver returned a value"
        )

    raise MissingSecret(f"Missing secret: {name}")

clear_secret_resolver()

Testing helper to reset resolver state.

Source code in octodns_metaname/secrets.py 
75
76
77
78
79
80
def clear_secret_resolver() -> None:
    """Testing helper to reset resolver state."""

    global _secret_resolver, _resolver_loaded
    _secret_resolver = None
    _resolver_loaded = False

Testing Helpers

Helpers used in the octodns-metaname test suite.

Test Suite

Client

Provider

Secrets