1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
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
55
56
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
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
|
import httpx
import asyncio
import sys
import json
from typing import Dict, Any, Optional
from tenacity import retry, stop_after_attempt, wait_exponential, retry_if_exception_type
class PorkbunAPI:
BASE_URL = "https://api.porkbun.com/api/json/v3"
def __init__(self, api_key: str, secret_key: str, debug: bool = False):
self.api_key = api_key
self.secret_key = secret_key
self.debug = debug
self.client = httpx.AsyncClient(timeout=30.0)
async def _make_request(self, endpoint: str, data: Optional[Dict[str, Any]] = None) -> Dict[str, Any]:
payload = {
"apikey": self.api_key,
"secretapikey": self.secret_key
}
if data:
payload.update(data)
url = f"{self.BASE_URL}/{endpoint}"
if self.debug:
print("=" * 80, file=sys.stderr)
print(f"DEBUG: REQUEST", file=sys.stderr)
print(f"URL: POST {url}", file=sys.stderr)
print(f"Headers: {dict(self.client.headers)}", file=sys.stderr)
print(f"Payload: {json.dumps(payload, indent=2)}", file=sys.stderr)
print("=" * 80, file=sys.stderr)
while True:
response = await self.client.post(url, json=payload)
if self.debug:
print("=" * 80, file=sys.stderr)
print(f"DEBUG: RESPONSE", file=sys.stderr)
print(f"Status: {response.status_code} {response.reason_phrase}", file=sys.stderr)
print(f"Headers: {dict(response.headers)}", file=sys.stderr)
print(f"Body: {response.text[:1000]}", file=sys.stderr)
if len(response.text) > 1000:
print(f"... (truncated, total length: {len(response.text)})", file=sys.stderr)
print("=" * 80, file=sys.stderr)
if response.status_code == 429:
retry_after = int(response.headers.get("Retry-After", 1))
if self.debug:
print(f"DEBUG: Rate limited, waiting {retry_after}s", file=sys.stderr)
await asyncio.sleep(retry_after)
continue
response.raise_for_status()
result = response.json()
if result.get("status") == "ERROR":
raise Exception(f"API Error: {result.get('message', 'Unknown error')}")
return result
async def ping(self) -> Dict[str, Any]:
return await self._make_request("ping")
async def get_pricing(self) -> Dict[str, Any]:
return await self._make_request("pricing/get")
async def check_domain_availability(self, domain: str) -> Dict[str, Any]:
return await self._make_request(f"domain/checkDomain/{domain}")
async def get_domain_pricing(self, domain: str) -> Dict[str, Any]:
pricing_data = await self.get_pricing()
if "pricing" not in pricing_data:
raise Exception("Invalid pricing response from API")
parts = domain.split('.')
if len(parts) < 2:
raise ValueError(f"Invalid domain format: {domain}")
tld = '.'.join(parts[1:])
result = {
"domain": domain,
"tld": tld
}
# Get pricing info
if tld in pricing_data["pricing"]:
result["pricing"] = pricing_data["pricing"][tld]
else:
result["error"] = "TLD not found in pricing data"
return result
# Check domain availability
try:
availability = await self.check_domain_availability(domain)
if availability.get("status") == "SUCCESS":
response_data = availability.get("response", {})
result["available"] = response_data.get("avail") == "yes"
if response_data.get("premium") == "yes":
result["premium"] = True
else:
result["available"] = None
except Exception as e:
result["available"] = None
return result
async def close(self):
await self.client.aclose()
|
