3
3
4
4
import os
5
5
from dataclasses import dataclass
6
- from ipaddress import ip_address , IPv4Address
7
- from typing import Set , List , NamedTuple , Optional
6
+ from ipaddress import IPv4Address
7
+ from typing import NamedTuple , Optional
8
8
9
9
import boto3
10
10
11
- NameDomain = NamedTuple ("NameSubdomain " , [("name" , str ), ("subdomain" , str )])
12
- Infrastructure = NamedTuple ("Infrastructure" , [("clusters" , List ["Cluster" ]),
13
- ("servers" , List ["Server" ])])
11
+ NameDomain = NamedTuple ("NameDomain " , [("name" , str ), ("subdomain" , str )])
12
+ Infrastructure = NamedTuple ("Infrastructure" , [("clusters" , list ["Cluster" ]),
13
+ ("servers" , list ["Server" ])])
14
14
15
15
CLUSTER_MAP = {1 : NameDomain ("Los Angeles" , "la" ),
16
16
2 : NameDomain ("New York" , "nyc" ),
@@ -57,7 +57,7 @@ class Cluster:
57
57
def __init__ (self , cluster_id : int , subdomain : str ) -> None :
58
58
self .cluster_id = cluster_id
59
59
self .subdomain = subdomain
60
- self .server_instances : List [Server ] = []
60
+ self .server_instances : list [Server ] = []
61
61
62
62
@property
63
63
def cluster_name (self ) -> str :
@@ -92,7 +92,7 @@ def _get_hosted_zone(self) -> dict:
92
92
raise RuntimeError (f"Failed to find hosted zone for { self .name } " )
93
93
94
94
@property
95
- def records (self ) -> List [dict ]:
95
+ def records (self ) -> list [dict ]:
96
96
"""Flexible record property"""
97
97
record_sets = self .r53 \
98
98
.list_resource_record_sets (HostedZoneId = self .id ) \
@@ -101,17 +101,17 @@ def records(self) -> List[dict]:
101
101
if (r ["Type" ] == "A" and r ["Name" ] != self .name )]
102
102
103
103
@staticmethod
104
- def ips_from_record (record : dict ) -> Set [IPv4Address ]:
104
+ def ips_from_record (record : dict ) -> set [IPv4Address ]:
105
105
"""Helper to get IP addresses from a given record"""
106
106
ips = set ()
107
107
for value in record ["ResourceRecords" ]:
108
- ips .add (ip_address (value ["Value" ]))
108
+ ips .add (IPv4Address (value ["Value" ]))
109
109
return ips
110
110
111
- def _a_record (self , name : str , ips : Set [IPv4Address ],
111
+ def _a_record (self , name : str , ips : set [IPv4Address ],
112
112
action : str , ttl : int = 30 ) -> None :
113
113
change_batch = {
114
- "Comment" : "add {} -> {}"
114
+ "Comment" : "add f {} -> {}" # pylint: disable=consider-using-f-string
115
115
.format (name , ", " .join ([ip .exploded for ip in ips ])),
116
116
"Changes" : [
117
117
{
@@ -134,7 +134,7 @@ def add_server(self, server: Server) -> None:
134
134
"""Adds the server's IP to the cluster's subdomain."""
135
135
fqdn = CLUSTER_MAP [server .cluster_id ].subdomain + "." + self .name
136
136
137
- ips : Set [IPv4Address ] = set ()
137
+ ips : set [IPv4Address ] = set ()
138
138
for record in self .records :
139
139
if fqdn == record ["Name" ]:
140
140
ips = self .ips_from_record (record )
@@ -149,7 +149,7 @@ def remove_server(self, server: Server) -> None:
149
149
"""Removes the server's IP from the DNS record."""
150
150
fqdn = CLUSTER_MAP [server .cluster_id ].subdomain + "." + self .name
151
151
152
- ips : Set [IPv4Address ] = set ()
152
+ ips : set [IPv4Address ] = set ()
153
153
for record in self .records :
154
154
if fqdn == record ["Name" ]:
155
155
ips = self .ips_from_record (record )
@@ -174,15 +174,15 @@ def create_infrastructure() -> Infrastructure:
174
174
for cluster_id , ident in CLUSTER_MAP .items ():
175
175
clusters .append (Cluster (cluster_id , ident .subdomain ))
176
176
177
- clusters [0 ].create_server (1 , ip = ip_address ("2.4.6.8" ))
178
- clusters [1 ].create_server (2 , ip = ip_address ("1.0.1.1" ))
179
- clusters [2 ].create_server (3 , ip = ip_address ("5.6.7.8" ))
180
- clusters [3 ].create_server (4 , ip = ip_address ("4.3.2.1" ))
181
- clusters [3 ].create_server (5 , ip = ip_address ("1.2.3.4" ))
182
- clusters [3 ].create_server (6 , ip = ip_address ("1.2.3.5" ))
183
- clusters [3 ].create_server (7 , ip = ip_address ("1.2.3.6" ))
184
- clusters [4 ].create_server (8 , ip = ip_address ("8.1.1.1" ))
185
- clusters [5 ].create_server (9 , ip = ip_address ("9.1.1.1" ))
177
+ clusters [0 ].create_server (1 , ip = IPv4Address ("2.4.6.8" ))
178
+ clusters [1 ].create_server (2 , ip = IPv4Address ("1.0.1.1" ))
179
+ clusters [2 ].create_server (3 , ip = IPv4Address ("5.6.7.8" ))
180
+ clusters [3 ].create_server (4 , ip = IPv4Address ("4.3.2.1" ))
181
+ clusters [3 ].create_server (5 , ip = IPv4Address ("1.2.3.4" ))
182
+ clusters [3 ].create_server (6 , ip = IPv4Address ("1.2.3.5" ))
183
+ clusters [3 ].create_server (7 , ip = IPv4Address ("1.2.3.6" ))
184
+ clusters [4 ].create_server (8 , ip = IPv4Address ("8.1.1.1" ))
185
+ clusters [5 ].create_server (9 , ip = IPv4Address ("9.1.1.1" ))
186
186
187
187
servers = []
188
188
for cluster in clusters :
@@ -193,15 +193,15 @@ def create_infrastructure() -> Infrastructure:
193
193
194
194
return Infrastructure (clusters , servers )
195
195
196
- def print_servers (server_instances : List [Server ]) -> None :
196
+ def print_servers (server_instances : list [Server ]) -> None :
197
197
"""ASCII analogy of the server UI"""
198
198
print ("\n \033 [1mID\t Name\t Cluster\t IP\t \t DNS\033 [0m" )
199
199
for server in server_instances :
200
200
print (server .server_id , "\t " , server .name , "\t " ,
201
201
CLUSTER_MAP [server .cluster_id ].name , "\t " ,
202
202
server .ip_address , server .dns )
203
203
204
- def print_dns (dns_records : List [dict ], server_instances : List [Server ]) -> None :
204
+ def print_dns (dns_records : list [dict ], server_instances : list [Server ]) -> None :
205
205
"""ASCII analogy of the DNS UI"""
206
206
print ("\n \033 [1mDomain\t \t \t \t \t IP(s)\t \t Server(s)\t Cluster\033 [0m" )
207
207
for record in dns_records :
@@ -214,7 +214,7 @@ def print_dns(dns_records: List[dict], server_instances: List[Server]) -> None:
214
214
[CLUSTER_MAP [server .cluster_id ].name
215
215
for server in matching_servers ])
216
216
217
- def update_servers (records : List [dict ], servers : List [Server ]) -> None :
217
+ def update_servers (records : list [dict ], servers : list [Server ]) -> None :
218
218
"""Assigns to each server instance their DNS record name"""
219
219
for server in servers :
220
220
server .dns = None
0 commit comments