HTTP Clients

HTTP client implementation for QLAM API communication.

HTTP clients for qlam-core.

This module provides thin wrappers around httpx that mirror the public API of qlam_client while integrating with qlam-core's auth providers.

Autogeneration and OpenAPI linkage: - Resource-specific plugins are backed by specs generated from OpenAPI definitions. Those specs are consumed by qlam_core.clients.http.universal_executor.execute_from_spec, which uses these clients to perform requests. This file itself is hand-written but works in concert with generated resource specs.

AuthenticatedClient

AuthenticatedClient(base_url: str, auth_provider: BaseAuthProvider, timeout: float | None = None, verify_ssl: bool | str = True, follow_redirects: bool = True, headers: Dict[str, str] | None = None, cookies: Dict[str, str] | None = None, logger: Logger | None = None)

Bases: Client

              flowchart TD
              qlam_core.clients.http.client.AuthenticatedClient[AuthenticatedClient]
              qlam_core.clients.http.client.Client[Client]

                              qlam_core.clients.http.client.Client --> qlam_core.clients.http.client.AuthenticatedClient
                


              click qlam_core.clients.http.client.AuthenticatedClient href "" "qlam_core.clients.http.client.AuthenticatedClient"
              click qlam_core.clients.http.client.Client href "" "qlam_core.clients.http.client.Client"
            

HTTP client with authentication support.

Extends Client with authentication capabilities by integrating with auth providers that implement HTTPHeadersMixin. The auth provider is responsible for converting whatever authentication method it uses (tokens, credentials, etc.) into appropriate HTTP headers.

This design keeps the HTTP client focused purely on HTTP concerns while allowing auth providers to handle the specifics of their authentication method.

:param base_url: Base URL for API requests. :param auth_provider: Auth provider (should implement HTTPHeadersMixin to provide authentication headers). :param timeout: Request timeout in seconds. :param verify_ssl: SSL verification (True, False, or path to CA bundle). :param follow_redirects: Whether to follow HTTP redirects. :param headers: Additional headers to include with requests. :param cookies: Default cookies to include with requests. :param logger: Logger instance for debugging. :raises ValueError: If auth_provider is None.

Source code in qlam_core/clients/http/client.py

def __init__(
    self,
    base_url: str,
    auth_provider: BaseAuthProvider,
    timeout: float | None = None,
    verify_ssl: bool | str = True,
    follow_redirects: bool = True,
    headers: Dict[str, str] | None = None,
    cookies: Dict[str, str] | None = None,
    logger: logging.Logger | None = None,
) -> None:
    """Initialize the authenticated HTTP client.

    :param base_url: Base URL for API requests.
    :param auth_provider: Auth provider (should implement HTTPHeadersMixin to provide authentication headers).
    :param timeout: Request timeout in seconds.
    :param verify_ssl: SSL verification (True, False, or path to CA bundle).
    :param follow_redirects: Whether to follow HTTP redirects.
    :param headers: Additional headers to include with requests.
    :param cookies: Default cookies to include with requests.
    :param logger: Logger instance for debugging.
    :raises ValueError: If `auth_provider` is None.
    """
    if auth_provider is None:
        raise ValueError("auth_provider is required")

    self._auth_provider = auth_provider
    auth_headers = self._get_auth_headers()
    conflicts = _find_header_conflicts_case_insensitive(headers, auth_headers)
    if conflicts:
        conflict_list = ", ".join(sorted(conflicts, key=str.lower))
        raise ConfigurationError(
            "The following headers are managed by auth_provider and cannot be set via "
            f"default headers: {conflict_list}. Use auth_provider instead."
        )

    # Initialize parent first to set up _logger
    super().__init__(
        base_url=base_url,
        timeout=timeout,
        verify_ssl=verify_ssl,
        follow_redirects=follow_redirects,
        headers=headers,
        cookies=cookies,
        logger=logger,
    )

    # Auth headers must win over SDK defaults and other non-auth headers.
    self._default_headers = _merge_headers_case_insensitive(self._default_headers, auth_headers)

auth_provider property

auth_provider: BaseAuthProvider

Get the authentication provider.

:return: Underlying auth provider instance.

refresh_auth_headers

refresh_auth_headers() -> None

Refresh authentication headers from the provider.

This updates default headers with fresh authentication information from the auth provider. Useful when tokens may have been refreshed.

Source code in qlam_core/clients/http/client.py

def refresh_auth_headers(self) -> None:
    """Refresh authentication headers from the provider.

    This updates default headers with fresh authentication information from
    the auth provider. Useful when tokens may have been refreshed.
    """
    auth_headers = self._get_auth_headers()

    # Update default headers, preserving non-auth headers.
    self._default_headers = _merge_headers_case_insensitive(self._default_headers, auth_headers)

    # Close existing clients to force recreation with new headers
    if self._httpx_client is not None:
        self._httpx_client.close()
        self._httpx_client = None

    if self._async_httpx_client is not None:
        # Note: Cannot await in sync method
        self._async_httpx_client = None
        self._logger.warning(
            "Async httpx client reset - will be recreated with new auth headers"
        )

with_cookies

with_cookies(cookies: Dict[str, str]) -> AuthenticatedClient

Return a new authenticated client with additional cookies.

:param cookies: Cookies to merge into defaults. :return: A new AuthenticatedClient instance.

Source code in qlam_core/clients/http/client.py

def with_cookies(self, cookies: Dict[str, str]) -> AuthenticatedClient:
    """Return a new authenticated client with additional cookies.

    :param cookies: Cookies to merge into defaults.
    :return: A new ``AuthenticatedClient`` instance.
    """
    merged_cookies = {**self._default_cookies, **cookies}
    auth_headers = self._get_auth_headers()
    return AuthenticatedClient(
        base_url=self._base_url,
        auth_provider=self._auth_provider,
        timeout=self._timeout,
        verify_ssl=self._verify_ssl,
        follow_redirects=self._follow_redirects,
        headers=_remove_headers_case_insensitive(self._default_headers, auth_headers),
        cookies=merged_cookies,
        logger=self._logger,
    )

with_headers

with_headers(headers: Dict[str, str]) -> AuthenticatedClient

Return a new authenticated client with additional headers.

:param headers: Headers to merge with fresh auth headers. :return: A new AuthenticatedClient instance.

Source code in qlam_core/clients/http/client.py

def with_headers(self, headers: Dict[str, str]) -> AuthenticatedClient:
    """Return a new authenticated client with additional headers.

    :param headers: Headers to merge with fresh auth headers.
    :return: A new ``AuthenticatedClient`` instance.
    """
    auth_headers = self._get_auth_headers()
    conflicts = _find_header_conflicts_case_insensitive(headers, auth_headers)
    if conflicts:
        conflict_list = ", ".join(sorted(conflicts, key=str.lower))
        raise ConfigurationError(
            "The following headers are managed by auth_provider and cannot be set via "
            f"default headers: {conflict_list}. Use auth_provider instead."
        )
    merged_headers = _merge_headers_case_insensitive(
        _remove_headers_case_insensitive(self._default_headers, auth_headers),
        headers,
    )

    return AuthenticatedClient(
        base_url=self._base_url,
        auth_provider=self._auth_provider,
        timeout=self._timeout,
        verify_ssl=self._verify_ssl,
        follow_redirects=self._follow_redirects,
        headers=merged_headers,
        cookies=self._default_cookies,
        logger=self._logger,
    )

with_timeout

with_timeout(timeout: float) -> AuthenticatedClient

Return a new authenticated client with a different timeout.

:param timeout: New timeout value in seconds. :return: A new AuthenticatedClient instance with timeout applied.

Source code in qlam_core/clients/http/client.py

def with_timeout(self, timeout: float) -> AuthenticatedClient:
    """Return a new authenticated client with a different timeout.

    :param timeout: New timeout value in seconds.
    :return: A new ``AuthenticatedClient`` instance with timeout applied.
    """
    auth_headers = self._get_auth_headers()
    return AuthenticatedClient(
        base_url=self._base_url,
        auth_provider=self._auth_provider,
        timeout=timeout,
        verify_ssl=self._verify_ssl,
        follow_redirects=self._follow_redirects,
        headers=_remove_headers_case_insensitive(self._default_headers, auth_headers),
        cookies=self._default_cookies,
        logger=self._logger,
    )

Client

Client(base_url: str, timeout: float | None = None, verify_ssl: bool | str = True, follow_redirects: bool = True, headers: Dict[str, str] | None = None, cookies: Dict[str, str] | None = None, logger: Logger | None = None)

HTTP client without authentication, mirroring qlam_client.Client API.

This client provides the same interface as qlam_client.Client but is designed for the QSH Client architecture. It requires explicit base_url configuration and fails fast if not properly configured.

:param base_url: Base URL for API requests (required, fails fast if not set). :param timeout: Request timeout in seconds. :param verify_ssl: SSL verification (True, False, or path to CA bundle). :param follow_redirects: Whether to follow HTTP redirects. :param headers: Default headers to include with requests. :param cookies: Default cookies to include with requests. :param logger: Logger instance for debugging. :raises ValueError: If base_url is empty or None.

Source code in qlam_core/clients/http/client.py

def __init__(
    self,
    base_url: str,
    timeout: float | None = None,
    verify_ssl: bool | str = True,
    follow_redirects: bool = True,
    headers: Dict[str, str] | None = None,
    cookies: Dict[str, str] | None = None,
    logger: logging.Logger | None = None,
) -> None:
    """Initialize the HTTP client.

    :param base_url: Base URL for API requests (required, fails fast if not set).
    :param timeout: Request timeout in seconds.
    :param verify_ssl: SSL verification (True, False, or path to CA bundle).
    :param follow_redirects: Whether to follow HTTP redirects.
    :param headers: Default headers to include with requests.
    :param cookies: Default cookies to include with requests.
    :param logger: Logger instance for debugging.
    :raises ValueError: If base_url is empty or None.
    """
    if not base_url:
        raise ValueError("base_url is required and cannot be empty")

    self._base_url = base_url.rstrip("/")
    self._timeout = timeout or 30.0
    self._verify_ssl = verify_ssl
    self._follow_redirects = follow_redirects
    self._default_headers = _merge_headers_case_insensitive({}, headers)
    self._default_cookies = cookies or {}
    self._logger = logger or logging.getLogger(__name__)

    # httpx client instances (sync and async)
    self._httpx_client: httpx.Client | None = None
    self._async_httpx_client: httpx.AsyncClient | None = None

    # Normalize core defaults onto canonical header names.
    accept = _pop_header_case_insensitive(self._default_headers, "Accept")
    self._default_headers["Accept"] = accept or "application/json"
    user_agent = _pop_header_case_insensitive(self._default_headers, "User-Agent")
    self._default_headers["User-Agent"] = _compose_user_agent(user_agent)

base_url property

base_url: str

Get the base URL.

:return: Base URL configured for the client.

__aenter__ async

__aenter__() -> Client

Enter async context manager.

:return: Self.

Source code in qlam_core/clients/http/client.py

async def __aenter__(self) -> Client:
    """Enter async context manager.

    :return: Self.
    """
    return self

__aexit__ async

__aexit__(exc_type: Type[BaseException] | None, exc_val: BaseException | None, exc_tb: TracebackType | None) -> None

Exit async context manager.

:param exc_type: Exception type. :param exc_val: Exception value. :param exc_tb: Exception traceback.

Source code in qlam_core/clients/http/client.py

async def __aexit__(
    self,
    exc_type: Type[BaseException] | None,
    exc_val: BaseException | None,
    exc_tb: TracebackType | None,
) -> None:
    """Exit async context manager.

    :param exc_type: Exception type.
    :param exc_val: Exception value.
    :param exc_tb: Exception traceback.
    """
    await self.aclose()

__enter__

__enter__() -> Client

Enter sync context manager.

:return: Self.

Source code in qlam_core/clients/http/client.py

def __enter__(self) -> Client:
    """Enter sync context manager.

    :return: Self.
    """
    return self

__exit__

__exit__(exc_type: Type[BaseException] | None, exc_val: BaseException | None, exc_tb: TracebackType | None) -> None

Exit sync context manager.

:param exc_type: Exception type. :param exc_val: Exception value. :param exc_tb: Exception traceback.

Source code in qlam_core/clients/http/client.py

def __exit__(
    self,
    exc_type: Type[BaseException] | None,
    exc_val: BaseException | None,
    exc_tb: TracebackType | None,
) -> None:
    """Exit sync context manager.

    :param exc_type: Exception type.
    :param exc_val: Exception value.
    :param exc_tb: Exception traceback.
    """
    self.close()

aclose async

aclose() -> None

Close both sync and async httpx clients (async version).

Source code in qlam_core/clients/http/client.py

async def aclose(self) -> None:
    """Close both sync and async httpx clients (async version)."""
    if self._httpx_client is not None:
        self._httpx_client.close()
        self._httpx_client = None

    if self._async_httpx_client is not None:
        await self._async_httpx_client.aclose()
        self._async_httpx_client = None

close

close() -> None

Close both sync and async httpx clients.

Source code in qlam_core/clients/http/client.py

def close(self) -> None:
    """Close both sync and async httpx clients."""
    if self._httpx_client is not None:
        self._httpx_client.close()
        self._httpx_client = None

    if self._async_httpx_client is not None:
        # Note: Cannot await in sync method
        # Users should call aclose() separately for async client
        self._async_httpx_client = None
        self._logger.warning("Async httpx client not closed - call aclose() for proper cleanup")

get_async_httpx_client

get_async_httpx_client() -> httpx.AsyncClient

Get the underlying httpx.AsyncClient instance.

:return: The configured httpx.AsyncClient.

Source code in qlam_core/clients/http/client.py

def get_async_httpx_client(self) -> httpx.AsyncClient:
    """Get the underlying httpx.AsyncClient instance.

    :return: The configured ``httpx.AsyncClient``.
    """
    if self._async_httpx_client is None:
        self._async_httpx_client = httpx.AsyncClient(
            base_url=self._base_url,
            headers=self._default_headers,
            cookies=self._default_cookies,
            timeout=self._timeout,
            verify=self._verify_ssl,
            follow_redirects=self._follow_redirects,
        )
        self._logger.debug(f"Created httpx.AsyncClient for {self._base_url}")
    return self._async_httpx_client

get_httpx_client

get_httpx_client() -> httpx.Client

Get the underlying httpx.Client instance.

:return: The configured httpx.Client.

Source code in qlam_core/clients/http/client.py

def get_httpx_client(self) -> httpx.Client:
    """Get the underlying httpx.Client instance.

    :return: The configured ``httpx.Client``.
    """
    if self._httpx_client is None:
        self._httpx_client = httpx.Client(
            base_url=self._base_url,
            headers=self._default_headers,
            cookies=self._default_cookies,
            timeout=self._timeout,
            verify=self._verify_ssl,
            follow_redirects=self._follow_redirects,
        )
        self._logger.debug(f"Created httpx.Client for {self._base_url}")
    return self._httpx_client

set_async_httpx_client

set_async_httpx_client(client: AsyncClient) -> None

Set a custom httpx.AsyncClient instance.

:param client: External httpx.AsyncClient to set.

Source code in qlam_core/clients/http/client.py

def set_async_httpx_client(self, client: httpx.AsyncClient) -> None:
    """Set a custom httpx.AsyncClient instance.

    :param client: External ``httpx.AsyncClient`` to set.
    """
    if self._async_httpx_client is not None:
        # Note: In async context, we can't await close() here
        # Users should ensure proper cleanup
        self._async_httpx_client = None
    self._async_httpx_client = client

set_httpx_client

set_httpx_client(client: Client) -> None

Set a custom httpx.Client instance.

:param client: External httpx.Client to set.

Source code in qlam_core/clients/http/client.py

def set_httpx_client(self, client: httpx.Client) -> None:
    """Set a custom httpx.Client instance.

    :param client: External ``httpx.Client`` to set.
    """
    if self._httpx_client is not None:
        self._httpx_client.close()
    self._httpx_client = client

with_cookies

with_cookies(cookies: Dict[str, str]) -> Client

Return a new client with additional cookies.

:param cookies: Cookies to merge into defaults. :return: A new Client instance with merged cookies.

Source code in qlam_core/clients/http/client.py

def with_cookies(self, cookies: Dict[str, str]) -> Client:
    """Return a new client with additional cookies.

    :param cookies: Cookies to merge into defaults.
    :return: A new ``Client`` instance with merged cookies.
    """
    merged_cookies = {**self._default_cookies, **cookies}
    return Client(
        base_url=self._base_url,
        timeout=self._timeout,
        verify_ssl=self._verify_ssl,
        follow_redirects=self._follow_redirects,
        headers=dict(self._default_headers),
        cookies=merged_cookies,
        logger=self._logger,
    )

with_headers

with_headers(headers: Dict[str, str]) -> Client

Return a new client with additional headers.

:param headers: Headers to merge into defaults. :return: A new Client instance with merged headers.

Source code in qlam_core/clients/http/client.py

def with_headers(self, headers: Dict[str, str]) -> Client:
    """Return a new client with additional headers.

    :param headers: Headers to merge into defaults.
    :return: A new ``Client`` instance with merged headers.
    """
    merged_headers = _merge_headers_case_insensitive(self._default_headers, headers)
    return Client(
        base_url=self._base_url,
        timeout=self._timeout,
        verify_ssl=self._verify_ssl,
        follow_redirects=self._follow_redirects,
        headers=merged_headers,
        cookies=self._default_cookies,
        logger=self._logger,
    )

with_timeout

with_timeout(timeout: float) -> Client

Return a new client with a different timeout.

:param timeout: New timeout value in seconds. :return: A new Client instance with the specified timeout.

Source code in qlam_core/clients/http/client.py

def with_timeout(self, timeout: float) -> Client:
    """Return a new client with a different timeout.

    :param timeout: New timeout value in seconds.
    :return: A new ``Client`` instance with the specified timeout.
    """
    return Client(
        base_url=self._base_url,
        timeout=timeout,
        verify_ssl=self._verify_ssl,
        follow_redirects=self._follow_redirects,
        headers=dict(self._default_headers),
        cookies=self._default_cookies,
        logger=self._logger,
    )