avwx.service.base
All service classes are based on the base Service class. Implementation is mostly left to the other high-level subclasses.
1""" 2All service classes are based on the base Service class. Implementation is mostly left to the other high-level subclasses. 3""" 4 5# pylint: disable=too-few-public-methods,unsubscriptable-object 6 7# stdlib 8from socket import gaierror 9from typing import Any, Optional, Tuple 10 11# library 12import httpx 13import httpcore 14 15# module 16from avwx.exceptions import SourceError 17 18 19_TIMEOUT_ERRORS = ( 20 httpx.ConnectTimeout, 21 httpx.ReadTimeout, 22 httpx.WriteTimeout, 23 httpx.PoolTimeout, 24 httpcore.ReadTimeout, 25 httpcore.WriteTimeout, 26 httpcore.PoolTimeout, 27) 28_CONNECTION_ERRORS = (gaierror, httpcore.ConnectError, httpx.ConnectError) 29_NETWORK_ERRORS = ( 30 httpcore.ReadError, 31 httpcore.NetworkError, 32 httpcore.RemoteProtocolError, 33) 34 35 36class Service: 37 """Base Service class for fetching reports""" 38 39 report_type: str 40 _url: str = "" 41 _valid_types: Tuple[str, ...] = tuple() 42 43 def __init__(self, report_type: str): 44 if self._valid_types and report_type not in self._valid_types: 45 raise ValueError( 46 f"'{report_type}' is not a valid report type for {self.__class__.__name__}. Expected {self._valid_types}" 47 ) 48 self.report_type = report_type 49 50 @property 51 def root(self) -> Optional[str]: 52 """Returns the service's root URL""" 53 if self._url is None: 54 return None 55 url = self._url[self._url.find("//") + 2 :] 56 return url[: url.find("/")] 57 58 59class CallsHTTP: 60 """Service mixin supporting HTTP requests""" 61 62 method: str = "GET" 63 64 async def _call( # pylint: disable=too-many-arguments 65 self, 66 url: str, 67 params: Optional[dict] = None, 68 headers: Optional[dict] = None, 69 data: Any = None, 70 timeout: int = 10, 71 retries: int = 3, 72 ) -> str: 73 name = self.__class__.__name__ 74 try: 75 async with httpx.AsyncClient( 76 timeout=timeout, 77 follow_redirects=True, 78 ) as client: 79 for _ in range(retries): 80 if self.method.lower() == "post": 81 resp = await client.post( 82 url, params=params, headers=headers, data=data 83 ) 84 else: 85 resp = await client.get(url, params=params, headers=headers) 86 if resp.status_code == 200: 87 break 88 # Skip retries if remote server error 89 if resp.status_code >= 500: 90 raise SourceError(f"{name} server returned {resp.status_code}") 91 else: 92 raise SourceError(f"{name} server returned {resp.status_code}") 93 except _TIMEOUT_ERRORS as timeout_error: 94 raise TimeoutError(f"Timeout from {name} server") from timeout_error 95 except _CONNECTION_ERRORS as connect_error: 96 raise ConnectionError( 97 f"Unable to connect to {name} server" 98 ) from connect_error 99 except _NETWORK_ERRORS as network_error: 100 raise ConnectionError( 101 f"Unable to read data from {name} server" 102 ) from network_error 103 return str(resp.text)
class
Service:
37class Service: 38 """Base Service class for fetching reports""" 39 40 report_type: str 41 _url: str = "" 42 _valid_types: Tuple[str, ...] = tuple() 43 44 def __init__(self, report_type: str): 45 if self._valid_types and report_type not in self._valid_types: 46 raise ValueError( 47 f"'{report_type}' is not a valid report type for {self.__class__.__name__}. Expected {self._valid_types}" 48 ) 49 self.report_type = report_type 50 51 @property 52 def root(self) -> Optional[str]: 53 """Returns the service's root URL""" 54 if self._url is None: 55 return None 56 url = self._url[self._url.find("//") + 2 :] 57 return url[: url.find("/")]
Base Service class for fetching reports
class
CallsHTTP:
60class CallsHTTP: 61 """Service mixin supporting HTTP requests""" 62 63 method: str = "GET" 64 65 async def _call( # pylint: disable=too-many-arguments 66 self, 67 url: str, 68 params: Optional[dict] = None, 69 headers: Optional[dict] = None, 70 data: Any = None, 71 timeout: int = 10, 72 retries: int = 3, 73 ) -> str: 74 name = self.__class__.__name__ 75 try: 76 async with httpx.AsyncClient( 77 timeout=timeout, 78 follow_redirects=True, 79 ) as client: 80 for _ in range(retries): 81 if self.method.lower() == "post": 82 resp = await client.post( 83 url, params=params, headers=headers, data=data 84 ) 85 else: 86 resp = await client.get(url, params=params, headers=headers) 87 if resp.status_code == 200: 88 break 89 # Skip retries if remote server error 90 if resp.status_code >= 500: 91 raise SourceError(f"{name} server returned {resp.status_code}") 92 else: 93 raise SourceError(f"{name} server returned {resp.status_code}") 94 except _TIMEOUT_ERRORS as timeout_error: 95 raise TimeoutError(f"Timeout from {name} server") from timeout_error 96 except _CONNECTION_ERRORS as connect_error: 97 raise ConnectionError( 98 f"Unable to connect to {name} server" 99 ) from connect_error 100 except _NETWORK_ERRORS as network_error: 101 raise ConnectionError( 102 f"Unable to read data from {name} server" 103 ) from network_error 104 return str(resp.text)
Service mixin supporting HTTP requests