"""Methods to for owsrequest.""" from typing import Any, Optional import httpx from owsclient import M2MTokenManager, OwsClient from secrets_manager.lambda_ext import LambdaSecretsManager # type: ignore 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, # extended timeout added to override owsclient default timeout for QA to finish long query, no impact on PROD timeout=httpx.Timeout(EXTENDED_TIMEOUT_SECONDS), ) def get(service: str, path: str, **options: Any) -> Any: """GET request to specified ows service.""" return ows_client.get(service, path, **options) def post(service: str, path: str, body: Any, **options: Any) -> Any: """POST request to specified ows service.""" return ows_client.post(service, path, json=body, **options) def put(service: str, path: str, body: Any, **options: Any) -> Any: """PUT request to specified ows service.""" return ows_client.put(service, path, json=body, **options) def delete(service: str, path: str, **options: Any) -> Any: """DELETE request to specified ows service.""" return ows_client.delete(service, path, **options)