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
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
|
import asyncio
import aiohttp
from typing import Dict, List, Optional
from pydantic import BaseModel
class TxtVerification(BaseModel):
name: str
token: str
class ZoneResponse(BaseModel):
id: str
created: str
modified: str
legacy_dns_host: str
legacy_ns: List[str]
name: str
ns: List[str]
owner: str
paused: bool
permission: str
project: str
registrar: str
status: str
ttl: int
verified: str
records_count: int
is_secondary_dns: bool
txt_verification: TxtVerification
class RecordResponse(BaseModel):
type: str
id: str
created: str
modified: str
zone_id: str
name: str
value: str
ttl: int | None = None
class HCloudDNS:
def __init__(self, token: str, api_domain="dns.hetzner.com"):
self._token = token
self.api_domain = api_domain
self._session: Optional[aiohttp.ClientSession] = None
def _get_session(self):
if self._session is None:
self._session = aiohttp.ClientSession(
headers={
"Auth-API-Token": self._token,
"Content-Type": "application/json; charset=utf-8",
}
)
return self._session
async def close(self):
if self._session is not None:
await self._session.close()
await asyncio.sleep(0.250)
async def get_all_zones(self, name: Optional[str] = None) -> List[ZoneResponse]:
session = self._get_session()
params = {}
if name is not None:
params["name"] = name
async with session.get(
f"https://{ self.api_domain }/api/v1/zones", params=params
) as resp:
resp.raise_for_status()
return [ZoneResponse(**zone) for zone in (await resp.json())["zones"]]
async def get_zone(self, zone_id: str) -> ZoneResponse:
session = self._get_session()
async with session.get(
f"https://{ self.api_domain }/api/v1/zones/{zone_id}"
) as resp:
resp.raise_for_status()
return ZoneResponse(**(await resp.json())["zone"])
async def get_zone_by_name(self, name: str) -> ZoneResponse:
zones = await self.get_all_zones(name)
if len(zones) == 0:
raise ValueError(f"Zone '{name}' not found")
elif len(zones) > 1:
raise ValueError(f"Multiple zones found for '{name}'")
else:
return zones[0]
async def get_all_records(self, zone_id: str) -> List[RecordResponse]:
session = self._get_session()
async with session.get(
f"https://{ self.api_domain }/api/v1/records", params={"zone_id": zone_id}
) as resp:
resp.raise_for_status()
data = await resp.json()
return [RecordResponse(**record) for record in data["records"]]
async def get_record(self, record_id: str) -> RecordResponse:
session = self._get_session()
async with session.get(
f"https://{ self.api_domain }/api/v1/records/{record_id}"
) as resp:
resp.raise_for_status()
return RecordResponse(**(await resp.json())["record"])
async def get_records_by_name(
self, zone_id: str, name: str
) -> List[RecordResponse]:
records = await self.get_all_records(zone_id)
return [record for record in records if record.name == name]
async def delete_record(self, record_id: str) -> None:
session = self._get_session()
async with session.delete(
f"https://{ self.api_domain }/api/v1/records/{record_id}"
) as resp:
resp.raise_for_status()
return None
async def create_record(
self,
zone_id: str,
name: str,
record_type: str,
value: str,
ttl: Optional[int] = None,
):
session = self._get_session()
data: Dict[str, str | int] = {
"name": name,
"type": record_type,
"value": value,
"zone_id": zone_id,
}
if ttl is not None:
data["ttl"] = ttl
async with session.post(
f"https://{ self.api_domain }/api/v1/records", json=data
) as resp:
resp.raise_for_status()
return RecordResponse(**(await resp.json())["record"])
|
