f8cd37ac8c
Includes: API (Express/TypeScript), Astro site, Python workers, document generators, FCC compliance tools, Canada CRTC formation, Ansible infrastructure, and deployment scripts. Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
122 lines
3.9 KiB
Python
122 lines
3.9 KiB
Python
"""
|
|
Porkbun API client for .ca domain operations.
|
|
|
|
Docs: https://porkbun.com/api/json/v3/documentation
|
|
All endpoints are POST with JSON body containing apikey + secretapikey.
|
|
"""
|
|
import logging
|
|
import os
|
|
from typing import Optional
|
|
|
|
import requests
|
|
|
|
LOG = logging.getLogger("tests.porkbun")
|
|
|
|
API_BASE = "https://api.porkbun.com/api/json/v3"
|
|
|
|
|
|
class PorkbunClient:
|
|
def __init__(
|
|
self,
|
|
api_key: Optional[str] = None,
|
|
secret_key: Optional[str] = None,
|
|
):
|
|
self.api_key = api_key or os.environ["PORKBUN_API_KEY"]
|
|
self.secret_key = secret_key or os.environ["PORKBUN_SECRET_KEY"]
|
|
self.session = requests.Session()
|
|
self.session.headers["Content-Type"] = "application/json"
|
|
|
|
def _auth_body(self, **extra) -> dict:
|
|
return {"apikey": self.api_key, "secretapikey": self.secret_key, **extra}
|
|
|
|
def ping(self) -> bool:
|
|
"""Verify API credentials are valid."""
|
|
r = self.session.post(f"{API_BASE}/ping", json=self._auth_body())
|
|
data = r.json()
|
|
if data.get("status") == "SUCCESS":
|
|
LOG.info("Porkbun API ping OK — IP: %s", data.get("yourIp"))
|
|
return True
|
|
LOG.error("Porkbun ping failed: %s", data)
|
|
return False
|
|
|
|
def check_availability(self, domain: str) -> dict:
|
|
"""Check if a domain is available for registration.
|
|
|
|
Returns: {"available": bool, "price": str, "currency": str}
|
|
"""
|
|
r = self.session.post(
|
|
f"{API_BASE}/domain/checkDomainAvailability/{domain}",
|
|
json=self._auth_body(),
|
|
timeout=15,
|
|
)
|
|
data = r.json()
|
|
avail = data.get("status") == "SUCCESS" and data.get("pricing")
|
|
price = ""
|
|
currency = ""
|
|
if avail and data.get("pricing"):
|
|
# Pricing is keyed by TLD
|
|
tld = domain.split(".", 1)[1] if "." in domain else ""
|
|
tld_pricing = data["pricing"].get(tld, {})
|
|
price = tld_pricing.get("registration", "")
|
|
currency = tld_pricing.get("currency", "USD")
|
|
|
|
return {
|
|
"available": avail,
|
|
"domain": domain,
|
|
"price": price,
|
|
"currency": currency,
|
|
"raw": data,
|
|
}
|
|
|
|
def register_domain(
|
|
self,
|
|
domain: str,
|
|
years: int = 1,
|
|
nameservers: Optional[list[str]] = None,
|
|
) -> dict:
|
|
"""Register a domain. Returns registration details."""
|
|
body = self._auth_body(domain=domain, years=years)
|
|
if nameservers:
|
|
body["ns"] = nameservers
|
|
|
|
r = self.session.post(
|
|
f"{API_BASE}/domain/register/{domain}",
|
|
json=body,
|
|
timeout=30,
|
|
)
|
|
data = r.json()
|
|
success = data.get("status") == "SUCCESS"
|
|
LOG.info("Porkbun register %s: %s", domain, "OK" if success else data.get("message"))
|
|
return {"success": success, "domain": domain, "raw": data}
|
|
|
|
def get_dns_records(self, domain: str) -> list[dict]:
|
|
"""List DNS records for a domain."""
|
|
r = self.session.post(
|
|
f"{API_BASE}/dns/retrieve/{domain}",
|
|
json=self._auth_body(),
|
|
timeout=10,
|
|
)
|
|
data = r.json()
|
|
return data.get("records", [])
|
|
|
|
def add_dns_record(
|
|
self, domain: str, record_type: str, name: str, content: str, ttl: int = 300
|
|
) -> dict:
|
|
"""Add a DNS record."""
|
|
r = self.session.post(
|
|
f"{API_BASE}/dns/create/{domain}",
|
|
json=self._auth_body(
|
|
type=record_type, name=name, content=content, ttl=str(ttl)
|
|
),
|
|
timeout=10,
|
|
)
|
|
return r.json()
|
|
|
|
def delete_domain(self, domain: str) -> dict:
|
|
"""Delete/cancel a domain (for test cleanup)."""
|
|
r = self.session.post(
|
|
f"{API_BASE}/domain/delete/{domain}",
|
|
json=self._auth_body(),
|
|
timeout=10,
|
|
)
|
|
return r.json()
|