33import logging
44from datetime import datetime , timedelta
55from functools import wraps
6- from typing import List , Optional , Tuple , Union
76from urllib .parse import parse_qs , urlencode , urljoin , urlsplit
87
98import requests
9+
10+ from ralph .assets .models import Service
1011from ralph .lib .dj_choices import Choices
1112from django .conf import settings
1213from django .utils .translation import gettext_lazy as _
1617
1718logger = logging .getLogger (__name__ )
1819
19- QueryParams = List [ Tuple [str , str ]]
20+ QueryParams = list [ tuple [str , str ]]
2021
2122
2223class RecordType (Choices ):
@@ -109,7 +110,7 @@ def _set_page_qp(url: str, page: int):
109110 result_url = _url ._replace (query = query ).geturl ()
110111 return result_url
111112
112- def get_api_result (self , url : str ) -> List [dict ]:
113+ def get_api_result (self , url : str ) -> list [dict ]:
113114 """
114115 Returns 'results' from DNSAAS API.
115116
@@ -129,13 +130,15 @@ def get_api_result(self, url: str) -> List[dict]:
129130 api_results .extend (_api_results )
130131 return api_results
131132
132- def _get_api_result (self , url : str ) -> Tuple [ List [dict ], bool ]:
133+ def _get_api_result (self , url : str ) -> tuple [ list [dict ], bool ]:
133134 status_code , json_data = self ._get (url )
134135 api_results = json_data .get ("content" , [])
135- last_page = bool (json_data .get ("last" , False ))
136+ last_page = bool (
137+ json_data .get ("last" , True )
138+ ) # if error, we don't want more pages
136139 return api_results , last_page
137140
138- def get_dns_records (self , ipaddresses : List [str ]) -> List [dict ]:
141+ def get_dns_records (self , ipaddresses : list [str ]) -> list [dict ]:
139142 """Gets DNS Records for `ipaddresses` by API call"""
140143 dns_records = []
141144 if not ipaddresses :
@@ -165,7 +168,7 @@ def get_dns_records(self, ipaddresses: List[str]) -> List[dict]:
165168 )
166169 return sorted (dns_records , key = lambda x : x ["type" ])
167170
168- def update_dns_record (self , record : dict ) -> Optional [ dict ] :
171+ def update_dns_record (self , record : dict ) -> dict | None :
169172 """
170173 Update DNS Record in DNSAAS
171174
@@ -187,7 +190,7 @@ def update_dns_record(self, record: dict) -> Optional[dict]:
187190 return response_data
188191
189192 @staticmethod
190- def _response2result (response : requests .Response ) -> Union [ dict , list , None ] :
193+ def _response2result (response : requests .Response ) -> dict | list | None :
191194 if response .status_code == 500 :
192195 logger .error ("Internal Server Error from DNSAAS: %s" , response .content )
193196 return {"non_field_errors" : ["Internal Server Error from DNSAAS" ]}
@@ -197,7 +200,7 @@ def _response2result(response: requests.Response) -> Union[dict, list, None]:
197200 except ValueError :
198201 pass
199202
200- def create_dns_record (self , record : dict , service = None ) -> Optional [ dict ] :
203+ def create_dns_record (self , record : dict , service : Service = None ) -> dict | None :
201204 """
202205 Create new DNS record.
203206
@@ -260,7 +263,7 @@ def _send_request_to_dnsaas(
260263 raise
261264
262265 @renew_token_when_unauthorized
263- def _post (self , url : str , data : dict ) -> [int , Optional [ dict ] ]:
266+ def _post (self , url : str , data : dict ) -> tuple [int , dict | None ]:
264267 """
265268 Send post data to URL.
266269
@@ -275,19 +278,19 @@ def _post(self, url: str, data: dict) -> [int, Optional[dict]]:
275278 return response .status_code , self ._response2result (response )
276279
277280 @renew_token_when_unauthorized
278- def _delete (self , url : str ) -> [int , Optional [ dict ] ]:
281+ def _delete (self , url : str ) -> tuple [int , dict | None ]:
279282 response = self ._send_request_to_dnsaas ("DELETE" , url )
280283
281284 return response .status_code , self ._response2result (response )
282285
283286 @renew_token_when_unauthorized
284- def _get (self , url : str ) -> [int , Optional [ dict ] ]:
287+ def _get (self , url : str ) -> tuple [int , dict | None ]:
285288 response = self ._send_request_to_dnsaas ("GET" , url )
286289
287290 return response .status_code , self ._response2result (response )
288291
289292 @renew_token_when_unauthorized
290- def _patch (self , url : str , data : dict ) -> [int , Optional [ dict ] ]:
293+ def _patch (self , url : str , data : dict ) -> tuple [int , dict | None ]:
291294 response = self ._send_request_to_dnsaas ("PATCH" , url , json_data = data )
292295 return response .status_code , self ._response2result (response )
293296
0 commit comments