Skip to content

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]):
    """Sync HTTPX backend."""

    __slots__ = ("_service_cache", "_client", "_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)

    @classmethod
    @override
    def bind_method(cls, signature: Signature) -> ServiceMethod[HttpxBackend]:  # noqa: D102
        response_type: TypeAdapter[Any] = get_type_adapter(cast(Hashable, 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 = self.__combadge_backend__._client.request(
                    request.get_method(),
                    request.get_url_path(),
                    json=request.payload,
                    data=request.form_data,
                    params=request.query_params,
                    headers=request.http_headers,
                )
                payload = self.__combadge_backend__._parse_payload(response)
            return signature.apply_response_markers(response, payload, response_type)

        return bound_method  # type: ignore[return-value]

    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 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)

Async

Async HTTPX backend.

Source code in combadge/support/httpx/backends/async_.py
class HttpxBackend(BaseHttpxBackend[AsyncClient]):
    """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)

    @classmethod
    @override
    def bind_method(cls, signature: Signature) -> ServiceMethod[HttpxBackend]:  # noqa: D102
        response_type: TypeAdapter[Any] = get_type_adapter(cast(Hashable, 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 self.__combadge_backend__._client.request(
                    request.get_method(),
                    request.get_url_path(),
                    json=request.payload,
                    data=request.form_data,
                    params=request.query_params,
                    headers=request.http_headers,
                )
                payload = self.__combadge_backend__._parse_payload(response)
            return signature.apply_response_markers(response, payload, response_type)

        return bound_method  # type: ignore[return-value]

    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 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)