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.

Initialize the authenticated HTTP client.

Parameters:

Name	Type	Description	Default
`base_url`	`str`	Base URL for API requests.	required
`auth_provider`	`BaseAuthProvider`	Auth provider (should implement HTTPHeadersMixin to provide authentication headers).	required
`timeout`	`float \| None`	Request timeout in seconds.	`None`
`verify_ssl`	`bool \| str`	SSL verification (True, False, or path to CA bundle).	`True`
`follow_redirects`	`bool`	Whether to follow HTTP redirects.	`True`
`headers`	`Dict[str, str] \| None`	Additional headers to include with requests.	`None`
`cookies`	`Dict[str, str] \| None`	Default cookies to include with requests.	`None`
`logger`	`Logger \| None`	Logger instance for debugging.	`None`

Raises:

Type	Description
`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.

Returns:

Type	Description
`BaseAuthProvider`	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.

Parameters:

Name	Type	Description	Default
`cookies`	`Dict[str, str]`	Cookies to merge into defaults.	required

Returns:

Type	Description
`AuthenticatedClient`	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.

Parameters:

Name	Type	Description	Default
`headers`	`Dict[str, str]`	Headers to merge with fresh auth headers.	required

Returns:

Type	Description
`AuthenticatedClient`	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.

Parameters:

Name	Type	Description	Default
`timeout`	`float`	New timeout value in seconds.	required

Returns:

Type	Description
`AuthenticatedClient`	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.

Initialize the HTTP client.

Parameters:

Name	Type	Description	Default
`base_url`	`str`	Base URL for API requests (required, fails fast if not set).	required
`timeout`	`float \| None`	Request timeout in seconds.	`None`
`verify_ssl`	`bool \| str`	SSL verification (True, False, or path to CA bundle).	`True`
`follow_redirects`	`bool`	Whether to follow HTTP redirects.	`True`
`headers`	`Dict[str, str] \| None`	Default headers to include with requests.	`None`
`cookies`	`Dict[str, str] \| None`	Default cookies to include with requests.	`None`
`logger`	`Logger \| None`	Logger instance for debugging.	`None`

Raises:

Type	Description
`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 = normalize_default_headers(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

base_url `property`

base_url: str

Get the base URL.

Returns:

Type	Description
`str`	Base URL configured for the client.

aenter `async`

__aenter__() -> Client

Enter async context manager.

Returns:

Type	Description
`Client`	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.

Parameters:

Name	Type	Description	Default
`exc_type`	`Type[BaseException] \| None`	Exception type.	required
`exc_val`	`BaseException \| None`	Exception value.	required
`exc_tb`	`TracebackType \| None`	Exception traceback.	required

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.

Returns:

Type	Description
`Client`	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.

Parameters:

Name	Type	Description	Default
`exc_type`	`Type[BaseException] \| None`	Exception type.	required
`exc_val`	`BaseException \| None`	Exception value.	required
`exc_tb`	`TracebackType \| None`	Exception traceback.	required

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.

Returns:

Type	Description
`AsyncClient`	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.

Returns:

Type	Description
`Client`	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.

Parameters:

Name	Type	Description	Default
`client`	`AsyncClient`	External `httpx.AsyncClient` to set.	required

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.

Parameters:

Name	Type	Description	Default
`client`	`Client`	External `httpx.Client` to set.	required

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.

Parameters:

Name	Type	Description	Default
`cookies`	`Dict[str, str]`	Cookies to merge into defaults.	required

Returns:

Type	Description
`Client`	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.

Parameters:

Name	Type	Description	Default
`headers`	`Dict[str, str]`	Headers to merge into defaults.	required

Returns:

Type	Description
`Client`	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.

Parameters:

Name	Type	Description	Default
`timeout`	`float`	New timeout value in seconds.	required

Returns:

Type	Description
`Client`	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,
    )

normalize_default_headers

normalize_default_headers(headers: Mapping[str, str] | None = None) -> Dict[str, str]

Apply the standard Accept and composed User-Agent headers.

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

def normalize_default_headers(
    headers: Mapping[str, str] | None = None,
) -> Dict[str, str]:
    """Apply the standard Accept and composed User-Agent headers."""
    normalized = _merge_headers_case_insensitive({}, dict(headers or {}))
    accept = _pop_header_case_insensitive(normalized, "Accept")
    normalized["Accept"] = accept or "application/json"
    user_agent = _pop_header_case_insensitive(normalized, "User-Agent")
    normalized["User-Agent"] = _compose_user_agent(user_agent)
    return normalized

HTTP Clients

AuthenticatedClient

auth_provider property

refresh_auth_headers

with_cookies

with_headers

with_timeout

Client

base_url property

__aenter__ async

__aexit__ async

__enter__

__exit__

aclose async

close

get_async_httpx_client

get_httpx_client

set_async_httpx_client

set_httpx_client

with_cookies

with_headers

with_timeout

normalize_default_headers

auth_provider `property`

base_url `property`

aenter `async`

aexit `async`

enter

exit

aclose `async`