Read API version and endpoint paths from settings

Author: qiyascc

CHANGELOG.md

@@ -11,3 +11,4 @@ First release.
 - Typed errors, with automatic retries for rate limits and server errors
 - Configuration through goodconf: code, `GODADDY_*` environment variables or a TOML, YAML or JSON file
 - Every user facing message lives in `messages.yaml` and can be overridden at runtime
+- API version and every endpoint path come from the settings and can be overridden

README.md

@@ -55,6 +55,8 @@ pip install "godaddy-python-sdk[yaml] @ git+https://github.com/qiyascc/godaddy-p
 | `base_url` | `GODADDY_BASE_URL` | taken from `environment` |
 | `production_url` | `GODADDY_PRODUCTION_URL` | `https://api.godaddy.com` |
 | `ote_url` | `GODADDY_OTE_URL` | `https://api.ote-godaddy.com` |
+| `api_version` | `GODADDY_API_VERSION` | `v1` |
+| `endpoints.*` | `GODADDY_ENDPOINTS__<NAME>` | see below |
 | `shopper_id` | `GODADDY_SHOPPER_ID` | |
 | `timeout` | `GODADDY_TIMEOUT` | `30` |
 | `max_retries` | `GODADDY_MAX_RETRIES` | `3` |
@@ -73,6 +75,34 @@ settings = load_settings("/etc/myapp/godaddy.toml", environment="ote")
 gd = GoDaddy(settings=settings)
 ```
 
+### Endpoints
+
+API paths are part of the settings too, so nothing is pinned in the code. Each one is a template relative to the base URL. `{version}` comes from `api_version`, and the other placeholders are filled in and URL encoded for each call.
+
+| Name | Default |
+| --- | --- |
+| `domains` | `/{version}/domains` |
+| `domain` | `/{version}/domains/{domain}` |
+| `available` | `/{version}/domains/available` |
+| `suggest` | `/{version}/domains/suggest` |
+| `tlds` | `/{version}/domains/tlds` |
+| `records` | `/{version}/domains/{domain}/records` |
+| `records_by_type` | `/{version}/domains/{domain}/records/{type}` |
+| `record` | `/{version}/domains/{domain}/records/{type}/{name}` |
+
+You can override one of them in a config file:
+
+```toml
+[endpoints]
+records = "/{version}/domains/{domain}/records"
+```
+
+or through the environment, with a double underscore between the parts:
+
+```bash
+export GODADDY_ENDPOINTS__TLDS="/{version}/domains/tlds"
+```
+
 GoDaddy gives you two environments. `production` is the default. `ote` is their test sandbox, and it needs its own keys.
 
 Heads up: GoDaddy only opens the production Domains API to some account types. If every call comes back with `PermissionDeniedError`, check your account's API access first.

src/godaddy/config.py

@@ -1,21 +1,34 @@
 from __future__ import annotations
 
 from typing import Any, Optional
+from urllib.parse import quote
 
 from goodconf import Field, GoodConf, GoodConfConfigDict
-from pydantic import SecretStr
+from pydantic import BaseModel, SecretStr
 
 from ._version import __version__
 from .errors import AuthenticationError
 from .messages import text
 
 
+class Endpoints(BaseModel):
+    domains: str = "/{version}/domains"
+    domain: str = "/{version}/domains/{domain}"
+    available: str = "/{version}/domains/available"
+    suggest: str = "/{version}/domains/suggest"
+    tlds: str = "/{version}/domains/tlds"
+    records: str = "/{version}/domains/{domain}/records"
+    records_by_type: str = "/{version}/domains/{domain}/records/{type}"
+    record: str = "/{version}/domains/{domain}/records/{type}/{name}"
+
+
 class Settings(GoodConf):
     model_config = GoodConfConfigDict(
         env_prefix="GODADDY_",
         file_env_var="GODADDY_CONFIG",
         default_files=["godaddy.toml", "godaddy.yaml", "godaddy.yml", "godaddy.json"],
         extra="ignore",
+        env_nested_delimiter="__",
     )
 
     api_key: SecretStr = Field(default=SecretStr(""), description=text("settings.api_key"))
@@ -24,6 +37,8 @@ class Settings(GoodConf):
     base_url: Optional[str] = Field(default=None, description=text("settings.base_url"))
     production_url: str = Field(default="https://api.godaddy.com", description=text("settings.production_url"))
     ote_url: str = Field(default="https://api.ote-godaddy.com", description=text("settings.ote_url"))
+    api_version: str = Field(default="v1", description=text("settings.api_version"))
+    endpoints: Endpoints = Field(default_factory=Endpoints, description=text("settings.endpoints"))
     shopper_id: Optional[str] = Field(default=None, description=text("settings.shopper_id"))
     timeout: float = Field(default=30.0, gt=0, description=text("settings.timeout"))
     max_retries: int = Field(default=3, ge=0, description=text("settings.max_retries"))
@@ -51,6 +66,11 @@ def resolved_base_url(self) -> str:
             raise ValueError(text("error.unknown_environment", environment=env, choices=choices))
         return self.environments[env].rstrip("/")
 
+    def path(self, endpoint: str, **params: str) -> str:
+        template = getattr(self.endpoints, endpoint)
+        values = {key: quote(str(value), safe="@") for key, value in params.items()}
+        return template.format(version=quote(self.api_version, safe=""), **values)
+
     def resolved_user_agent(self) -> str:
         return self.user_agent.replace("{version}", __version__)
 

src/godaddy/messages.yaml

@@ -6,6 +6,8 @@ settings:
   base_url: Full API URL. Overrides the environment when set.
   production_url: API URL used for the production environment.
   ote_url: API URL used for the OTE test environment.
+  api_version: API version that fills the {version} part of every endpoint.
+  endpoints: Path templates for every API call, relative to the base URL.
   shopper_id: Customer shopper id for reseller calls, sent as X-Shopper-Id.
   timeout: Seconds to wait for a single request.
   max_retries: How many times a failed request is tried again.

src/godaddy/resources/dns.py

@@ -2,7 +2,6 @@
 
 import ipaddress
 from typing import Any, Iterable, Optional, Sequence, Union
-from urllib.parse import quote
 
 from .._base import Call
 from ..messages import text
@@ -15,14 +14,14 @@ def _domain(domain: str) -> str:
     value = domain.strip().lower().rstrip(".")
     if not value:
         raise ValueError(text("value.empty_domain"))
-    return quote(value, safe="")
+    return value
 
 
 def _name(name: str) -> str:
     value = name.strip()
     if not value:
         raise ValueError(text("value.empty_name"))
-    return quote(value, safe="@")
+    return value
 
 
 def _records(payload: Any) -> list[DNSRecord]:
@@ -55,6 +54,9 @@ def __init__(self, client: Any) -> None:
     def _run(self, call: Call) -> Any:
         return self._client._execute(call)
 
+    def _path(self, endpoint: str, **params: str) -> str:
+        return self._client.settings.path(endpoint, **params)
+
     def list(
         self,
         domain: str,
@@ -64,15 +66,17 @@ def list(
         offset: Optional[int] = None,
         limit: Optional[int] = None,
     ) -> Any:
-        path = f"/v1/domains/{_domain(domain)}/records"
         wanted_name = name.strip() if name else None
 
         if type is not None:
-            path += f"/{quote(record_type(type), safe='')}"
             if wanted_name:
-                path += f"/{_name(wanted_name)}"
+                path = self._path("record", domain=_domain(domain), type=record_type(type), name=_name(wanted_name))
+            else:
+                path = self._path("records_by_type", domain=_domain(domain), type=record_type(type))
             parse = _records
         else:
+            path = self._path("records", domain=_domain(domain))
+
             def parse(payload: Any) -> list[DNSRecord]:
                 records = _records(payload)
                 if wanted_name:
@@ -82,18 +86,18 @@ def parse(payload: Any) -> list[DNSRecord]:
         return self._run(Call("GET", path, params={"offset": offset, "limit": limit}, parse=parse))
 
     def get(self, domain: str, type: RecordTypeLike, name: str) -> Any:
-        path = f"/v1/domains/{_domain(domain)}/records/{quote(record_type(type), safe='')}/{_name(name)}"
+        path = self._path("record", domain=_domain(domain), type=record_type(type), name=_name(name))
         return self._run(Call("GET", path, parse=_records))
 
     def exists(self, domain: str, type: RecordTypeLike, name: str) -> Any:
-        path = f"/v1/domains/{_domain(domain)}/records/{quote(record_type(type), safe='')}/{_name(name)}"
+        path = self._path("record", domain=_domain(domain), type=record_type(type), name=_name(name))
         return self._run(Call("GET", path, parse=lambda payload: bool(_records(payload))))
 
     def add(self, domain: str, *records: DNSRecord) -> Any:
         if not records:
             raise ValueError(text("value.add_needs_records"))
         body = [record.to_api() for record in records]
-        return self._run(Call("PATCH", f"/v1/domains/{_domain(domain)}/records", json=body, parse=lambda _: None))
+        return self._run(Call("PATCH", self._path("records", domain=_domain(domain)), json=body, parse=lambda _: None))
 
     def replace(
         self,
@@ -104,7 +108,7 @@ def replace(
         *,
         ttl: int = 600,
     ) -> Any:
-        path = f"/v1/domains/{_domain(domain)}/records/{quote(record_type(type), safe='')}/{_name(name)}"
+        path = self._path("record", domain=_domain(domain), type=record_type(type), name=_name(name))
         body = _record_bodies(values, ttl)
         return self._run(Call("PUT", path, json=body, parse=lambda _: None))
 
@@ -114,15 +118,15 @@ def replace_type(self, domain: str, type: RecordTypeLike, records: Sequence[DNSR
             if record.type != kind:
                 raise ValueError(text("value.type_mismatch", name=record.name, actual=record.type, expected=kind))
         body = [{"name": r.name, **r.to_api(include_identity=False)} for r in records]
-        path = f"/v1/domains/{_domain(domain)}/records/{quote(kind, safe='')}"
+        path = self._path("records_by_type", domain=_domain(domain), type=kind)
         return self._run(Call("PUT", path, json=body, parse=lambda _: None))
 
     def replace_all(self, domain: str, records: Sequence[DNSRecord]) -> Any:
         body = [record.to_api() for record in records]
-        return self._run(Call("PUT", f"/v1/domains/{_domain(domain)}/records", json=body, parse=lambda _: None))
+        return self._run(Call("PUT", self._path("records", domain=_domain(domain)), json=body, parse=lambda _: None))
 
     def delete(self, domain: str, type: RecordTypeLike, name: str, *, missing_ok: bool = False) -> Any:
-        path = f"/v1/domains/{_domain(domain)}/records/{quote(record_type(type), safe='')}/{_name(name)}"
+        path = self._path("record", domain=_domain(domain), type=record_type(type), name=_name(name))
         allow = frozenset({404}) if missing_ok else frozenset()
         return self._run(Call("DELETE", path, parse=lambda _: None, allow_status=allow))
 

src/godaddy/resources/domains.py

@@ -26,6 +26,9 @@ def __init__(self, client: Any) -> None:
     def _run(self, call: Call) -> Any:
         return self._client._execute(call)
 
+    def _path(self, endpoint: str, **params: str) -> str:
+        return self._client.settings.path(endpoint, **params)
+
     def _list_call(
         self,
         statuses: Optional[Iterable[str]],
@@ -41,7 +44,7 @@ def _list_call(
             "marker": marker,
             "includes": list(includes) if includes else None,
         }
-        return Call("GET", "/v1/domains", params=params, parse=_domains)
+        return Call("GET", self._path("domains"), params=params, parse=_domains)
 
     def list(
         self,
@@ -55,22 +58,22 @@ def list(
         return self._run(self._list_call(statuses, status_groups, limit, marker, includes))
 
     def get(self, domain: str) -> Any:
-        return self._run(Call("GET", f"/v1/domains/{_domain(domain)}", parse=Domain.from_api))
+        return self._run(Call("GET", self._path("domain", domain=_domain(domain)), parse=Domain.from_api))
 
     def available(self, domain: str, *, full_check: bool = False, for_transfer: bool = False) -> Any:
         params = {
             "domain": domain.strip().lower(),
             "checkType": "FULL" if full_check else "FAST",
             "forTransfer": for_transfer,
         }
-        return self._run(Call("GET", "/v1/domains/available", params=params, parse=Availability.from_api))
+        return self._run(Call("GET", self._path("available"), params=params, parse=Availability.from_api))
 
     def available_many(self, domains: Iterable[str], *, full_check: bool = False) -> Any:
         body = [d.strip().lower() for d in domains]
         if not body:
             raise ValueError(text("value.no_domains"))
         params = {"checkType": "FULL" if full_check else "FAST"}
-        return self._run(Call("POST", "/v1/domains/available", params=params, json=body, parse=_bulk_availability))
+        return self._run(Call("POST", self._path("available"), params=params, json=body, parse=_bulk_availability))
 
     def suggest(
         self,
@@ -95,11 +98,11 @@ def suggest(
             "waitMs": wait_ms,
         }
         parse = lambda payload: [Suggestion.from_api(item) for item in payload or []]
-        return self._run(Call("GET", "/v1/domains/suggest", params=params, parse=parse))
+        return self._run(Call("GET", self._path("suggest"), params=params, parse=parse))
 
     def tlds(self) -> Any:
         parse = lambda payload: [Tld.from_api(item) for item in payload or []]
-        return self._run(Call("GET", "/v1/domains/tlds", parse=parse))
+        return self._run(Call("GET", self._path("tlds"), parse=parse))
 
     def update(
         self,
@@ -121,7 +124,7 @@ def update(
             body["subaccountId"] = subaccount_id
         if not body:
             raise ValueError(text("value.update_needs_fields"))
-        return self._run(Call("PATCH", f"/v1/domains/{_domain(domain)}", json=body, parse=lambda _: None))
+        return self._run(Call("PATCH", self._path("domain", domain=_domain(domain)), json=body, parse=lambda _: None))
 
 
 class Domains(_DomainsBase):

tests/test_client.py

@@ -99,3 +99,40 @@ def test_headers_sent(gd, api):
     assert headers["authorization"] == "sso-key key:secret"
     assert headers["accept"] == "application/json"
     assert headers["user-agent"].startswith("godaddy-python-sdk/")
+
+
+def test_endpoints_from_settings(api):
+    import httpx
+
+    http = httpx.Client(transport=httpx.MockTransport(api))
+    client = GoDaddy("k", "s", http_client=http, api_version="v2")
+    api.queue(200, [])
+    client.dns.list("example.com", "A", "www")
+    assert api.last.url.path == "/v2/domains/example.com/records/A/www"
+
+
+def test_endpoint_override_from_env(monkeypatch, api):
+    import httpx
+
+    monkeypatch.setenv("GODADDY_ENDPOINTS__RECORDS", "/custom/{version}/zones/{domain}")
+    http = httpx.Client(transport=httpx.MockTransport(api))
+    client = GoDaddy("k", "s", http_client=http)
+    api.queue(200, [])
+    client.dns.list("example.com")
+    assert api.last.url.path == "/custom/v1/zones/example.com"
+
+
+def test_endpoint_override_from_file(monkeypatch, tmp_path):
+    (tmp_path / "godaddy.toml").write_text(
+        'api_key = "k"\napi_secret = "s"\n\n[endpoints]\ntlds = "/{version}/catalog/tlds"\n'
+    )
+    settings = load_settings()
+    assert settings.path("tlds") == "/v1/catalog/tlds"
+    assert settings.path("domains") == "/v1/domains"
+
+
+def test_path_values_are_quoted():
+    settings = load_settings(api_key="k", api_secret="s")
+    path = settings.path("record", domain="example.com", type="TXT", name="a b/c")
+    assert path == "/v1/domains/example.com/records/TXT/a%20b%2Fc"
+    assert settings.path("record", domain="example.com", type="A", name="@").endswith("/A/@")