from __future__ import annotations
import logging
from abc import ABC, abstractmethod
from typing import Any, Generic, Optional, TypeVar, Union, cast
import httpx
from imednet.auth.strategy import AuthStrategy
from imednet.constants import (
CONTENT_TYPE_JSON,
HEADER_ACCEPT,
HEADER_CONTENT_TYPE,
)
from imednet.core.http.executor import BaseRequestExecutor
from .base_client import BaseClient, Tracer
from .retry import RetryPolicy
logger = logging.getLogger(__name__)
ClientT = TypeVar("ClientT", bound=Union[httpx.Client, httpx.AsyncClient])
ExecutorT = TypeVar("ExecutorT", bound=BaseRequestExecutor)
[docs]class HTTPClientBase(BaseClient, ABC, Generic[ClientT, ExecutorT]):
"""Shared logic for synchronous and asynchronous HTTP clients."""
_client: ClientT
_executor: ExecutorT
@abstractmethod
def _get_client_class(self) -> type[ClientT]:
"""Return the underlying httpx client class."""
@abstractmethod
def _create_executor(
self, client: ClientT, retry_policy: Optional[RetryPolicy] = None
) -> ExecutorT:
"""Create the request executor."""
[docs] def __init__(
self,
api_key: Optional[str] = None,
security_key: Optional[str] = None,
base_url: Optional[str] = None,
timeout: Union[float, httpx.Timeout] = 30.0,
retries: int = 3,
backoff_factor: float = 1.0,
tracer: Optional[Tracer] = None,
retry_policy: RetryPolicy | None = None,
auth: Optional[AuthStrategy] = None,
) -> None:
super().__init__(
api_key=api_key,
security_key=security_key,
base_url=base_url,
timeout=timeout,
retries=retries,
backoff_factor=backoff_factor,
tracer=tracer,
auth=auth,
)
self._executor = self._create_executor(self._client, retry_policy)
def _create_client(self, auth: AuthStrategy) -> ClientT:
headers = {
HEADER_ACCEPT: CONTENT_TYPE_JSON,
HEADER_CONTENT_TYPE: CONTENT_TYPE_JSON,
}
headers.update(auth.get_headers())
client_cls = self._get_client_class()
return cast(
ClientT,
client_cls(
base_url=self._base_url,
headers=headers,
timeout=self.timeout,
),
)
@property
def retry_policy(self) -> RetryPolicy:
return cast(RetryPolicy, self._executor.retry_policy)
@retry_policy.setter
def retry_policy(self, policy: RetryPolicy) -> None:
self._executor.retry_policy = policy
@abstractmethod
def _request(self, method: str, path: str, **kwargs: Any) -> Any:
"""Make an HTTP request. Return type varies by async/sync client."""