HTTPX

Support for HTTPX client – requires combadge[httpx] extra.

Sync

Sync HTTPX backend.

Source code in combadge/support/httpx/backends/sync.py

class HttpxBackend(BaseHttpxBackend[Client], ServiceContainer):
    """Sync HTTPX backend."""

    __slots__ = ("_client", "_service_cache", "_raise_for_status")

    def __init__(
        self,
        client: Client,
        *,
        raise_for_status: bool = True,
    ) -> None:
        """
        Instantiate the backend.

        Args:
            client: [HTTPX client](https://www.python-httpx.org/advanced/#client-instances)
            raise_for_status: automatically call `raise_for_status()`
        """
        BaseHttpxBackend.__init__(self, client, raise_for_status=raise_for_status)
        ServiceContainer.__init__(self)

    def bind_method(self, signature: Signature) -> ServiceMethod[HttpxBackend]:  # noqa: D102
        backend = self
        response_type: TypeAdapter[Any] = TypeAdapter(signature.return_type)

        def bound_method(self: BaseBoundService[HttpxBackend], *args: Any, **kwargs: Any) -> Any:
            request = signature.build_request(Request, self, args, kwargs)
            with BackendError:
                response: Response = backend._client.request(
                    request.get_method(),
                    request.get_url_path(),
                    json=request.payload,
                    data=request.form_data,
                    params=request.query_params,
                    headers=request.headers,
                )
                payload = backend._parse_payload(response)
            return signature.apply_response_markers(response, payload, response_type)

        return bound_method  # type: ignore[return-value]

    binder = bind_method  # type: ignore[assignment]

    def __enter__(self) -> Self:
        self._client = self._client.__enter__()
        return self

    def __exit__(
        self,
        exc_type: type[BaseException] | None,
        exc_value: BaseException | None,
        traceback: TracebackType | None,
    ) -> Any:
        return self._client.__exit__(exc_type, exc_value, traceback)

__init__

__init__(client: Client, *, raise_for_status: bool = True) -> None

Instantiate the backend.

Parameters:

Name	Type	Description	Default
client	Client	HTTPX client	required
raise_for_status	bool	automatically call raise_for_status()	True

Source code in combadge/support/httpx/backends/sync.py

def __init__(
    self,
    client: Client,
    *,
    raise_for_status: bool = True,
) -> None:
    """
    Instantiate the backend.

    Args:
        client: [HTTPX client](https://www.python-httpx.org/advanced/#client-instances)
        raise_for_status: automatically call `raise_for_status()`
    """
    BaseHttpxBackend.__init__(self, client, raise_for_status=raise_for_status)
    ServiceContainer.__init__(self)

Async

Async HTTPX backend.

Source code in combadge/support/httpx/backends/async_.py

class HttpxBackend(BaseHttpxBackend[AsyncClient], ServiceContainer):
    """Async HTTPX backend."""

    __slots__ = ("_client", "_service_cache", "_raise_for_status")

    def __init__(
        self,
        client: AsyncClient,
        *,
        raise_for_status: bool = True,
    ) -> None:
        """
        Instantiate the backend.

        Args:
            client: [HTTPX client](https://www.python-httpx.org/advanced/#client-instances)
            raise_for_status: automatically call `raise_for_status()`
        """
        BaseHttpxBackend.__init__(self, client, raise_for_status=raise_for_status)
        ServiceContainer.__init__(self)

    def bind_method(self, signature: Signature) -> ServiceMethod[HttpxBackend]:  # noqa: D102
        backend = self
        response_type: TypeAdapter[Any] = TypeAdapter(signature.return_type)

        async def bound_method(self: BaseBoundService[HttpxBackend], *args: Any, **kwargs: Any) -> Any:
            request = signature.build_request(Request, self, args, kwargs)
            with BackendError:
                response: Response = await backend._client.request(
                    request.get_method(),
                    request.get_url_path(),
                    json=request.payload,
                    data=request.form_data,
                    params=request.query_params,
                    headers=request.headers,
                )
                payload = backend._parse_payload(response)
            return signature.apply_response_markers(response, payload, response_type)

        return bound_method  # type: ignore[return-value]

    binder = bind_method  # type: ignore[assignment]

    async def __aenter__(self) -> Self:
        self._client = await self._client.__aenter__()
        return self

    async def __aexit__(
        self,
        exc_type: type[BaseException] | None,
        exc_value: BaseException | None,
        traceback: TracebackType | None,
    ) -> Any:
        return await self._client.__aexit__(exc_type, exc_value, traceback)

__init__

__init__(client: AsyncClient, *, raise_for_status: bool = True) -> None

Instantiate the backend.

Parameters:

Name	Type	Description	Default
client	AsyncClient	HTTPX client	required
raise_for_status	bool	automatically call raise_for_status()	True

Source code in combadge/support/httpx/backends/async_.py

def __init__(
    self,
    client: AsyncClient,
    *,
    raise_for_status: bool = True,
) -> None:
    """
    Instantiate the backend.

    Args:
        client: [HTTPX client](https://www.python-httpx.org/advanced/#client-instances)
        raise_for_status: automatically call `raise_for_status()`
    """
    BaseHttpxBackend.__init__(self, client, raise_for_status=raise_for_status)
    ServiceContainer.__init__(self)