"""Methods for ows-client.""" from typing import Any, Optional from httpx import HTTPStatusError, Timeout from owsclient import M2MTokenManager, OwsClient from secrets_manager.lambda_ext import LambdaSecretsManager from config import APPLICATION_NAME, ENVIRONMENT from src.constants import EXTENDED_TIMEOUT_SECONDS secrets_manager = LambdaSecretsManager( environment=ENVIRONMENT, service_name=APPLICATION_NAME, ) m2m_token_manager = M2MTokenManager( secrets_manager=secrets_manager, environment=ENVIRONMENT, service_name=APPLICATION_NAME, ) ows_client = OwsClient( environment=ENVIRONMENT, service_name=APPLICATION_NAME, m2m_token_manager=m2m_token_manager, timeout=Timeout(EXTENDED_TIMEOUT_SECONDS), ) def delete(service: str, path: str, **options: Any) -> Any: """DELETE request to specified ows service.""" return _request(service, path, 'DELETE', **options) def get(service: str, path: str, **options: Any) -> Any: """GET request to specified ows service.""" return _request(service, path, 'GET', body=None, **options) def post(service: str, path: str, body: Any, **options: Any) -> Any: """POST request to specified ows service.""" return _request(service, path, 'POST', body, **options) def put(service: str, path: str, body: Any, **options: Any) -> Any: """PUT request to specified ows service.""" return _request(service, path, 'PUT', body, **options) def _request( service: str, path: str, method: str, body: Optional[Any] = None, raise_for_status: bool = False, **options: Any, ) -> Any: """Make request.""" response = ows_client.request(service, method, path, json=body, **options) if raise_for_status: try: response.raise_for_status() except HTTPStatusError as e: if e.response: raise HTTPStatusError( message=f'{e}: {e.response.text}', request=e.request, response=e.response, ) raise return response