from apollo_utils.service.exceptions import APIUnavailable from typing import Optional, Tuple, Type from server import config from server.client.exceptions import APIMisconfigured class HttpClientConfig: """Base configuration class for http service clients. Attributes: service: service name. uri: service base uri. timeout: requests timeout in seconds. passed_headers: tuple of names corresponding to headers that are only to be passed from the original request. If None all headers allowed are passed. retry_allowed: flag of allowing retries for this clients requests. retry_count: max attempts number for retrying. retry_delay: delay between retries in seconds. retry_excepted: tuple of exceptions that are only handled with retries, another exceptions are being risen. """ service: str uri: str timeout: int = config.DEFAULT_REQUEST_TIMEOUT passed_headers: Optional[Tuple[str]] = None retry_allowed: bool = True retry_count: int = config.DEFAULT_RETRY_COUNT retry_delay: int = config.DEFAULT_RETRY_WAIT retry_excepted: Tuple[Type[Exception]] = (APIUnavailable,) def __init__(self): self.check_not_nullable("uri") def check_not_nullable(self, arg_name: str): if getattr(self, arg_name, None) is None: raise APIMisconfigured(f"{self.service}: {arg_name} is not set.") class ApiKeyClientConfig(HttpClientConfig): """Base configuration class for http clients authenticated by authorization key. Attributes: auth_secret: authorization key. auth_name: header name to store auth_secret by. """ auth_secret: str auth_name: str = "Authorization" def __init__(self): super().__init__() self.check_not_nullable("auth_secret") class ClientConfig(HttpClientConfig): """ Base configuration class for http clients with no authorization key. """ pass