Metadata-Version: 2.5
Name: cloudru-qa-http-client
Version: 1.0.0
Summary: Базовый HTTP-клиент
Project-URL: Repository, https://github.com/cloud-ru-tech/qa-libraries
Project-URL: Issues, https://github.com/cloud-ru-tech/qa-libraries/issues
Author-email: Алексей Кирютин <ayukiryutin@cloud.ru>
License-Expression: Apache-2.0
License-File: LICENSE
Keywords: async,client,http,httpx
Classifier: Development Status :: 5 - Production/Stable
Classifier: Intended Audience :: Developers
Classifier: Operating System :: OS Independent
Classifier: Programming Language :: Python :: 3
Classifier: Programming Language :: Python :: 3.13
Classifier: Programming Language :: Python :: 3.14
Classifier: Topic :: Internet :: WWW/HTTP
Classifier: Topic :: Software Development :: Libraries :: Python Modules
Classifier: Typing :: Typed
Requires-Python: >=3.13
Requires-Dist: glom>=25.12.0
Requires-Dist: httpx<0.29,>=0.28.1
Provides-Extra: truststore
Requires-Dist: truststore<0.11,>=0.10.4; extra == 'truststore'
Description-Content-Type: text/markdown

# HTTP client

[![PyPI](https://img.shields.io/pypi/v/cloudru-qa-http-client)](https://pypi.org/project/cloudru-qa-http-client/)
[![Python](https://img.shields.io/pypi/pyversions/cloudru-qa-http-client)](https://pypi.org/project/cloudru-qa-http-client/)
[![License](https://img.shields.io/pypi/l/cloudru-qa-http-client)](https://github.com/cloud-ru-tech/qa-libraries/blob/master/LICENSE)

English | [Русский](README.ru.md)

- [HTTP client](#http-client)
  - [Installation](#installation)
  - [Skills for AI agents](#skills-for-ai-agents)
  - [Usage](#usage)
    - [Example:](#example)
      - [Base client example](#base-client-example)
      - [Custom client example](#custom-client-example)
      - [Asynchronous client example](#asynchronous-client-example)
    - [Configuration](#configuration)
  - [Available options](#available-options)
    - [EventHooks](#eventhooks)
    - [Decorating the send method](#decorating-the-send-method)
    - [Streaming requests](#streaming-requests)
      - [The `stream` method](#the-stream-method)
      - [The `stream_request` method](#the-stream_request-method)
      - [Decorating the stream\_request method](#decorating-the-stream_request-method)
    - [The request method](#the-request-method)
    - [Timeouts](#timeouts)
    - [Authorization](#authorization)
    - [Request identification](#request-identification)
    - [Tracing](#tracing)
    - [Logging](#logging)
    - [Closing connections](#closing-connections)
    - [TLS connection](#tls-connection)
    - [Default headers](#default-headers)
  - [Development](#development)

## Installation

Requires Python 3.13 or later.

```shell
pip install cloudru-qa-http-client
```

Import the library as the `http_client` module.

## Skills for AI agents

The package includes the [`http-client-usage`](https://github.com/cloud-ru-tech/qa-libraries/blob/master/.agents/skills/http-client-usage/SKILL.md) skill
with examples and usage rules for AI agents. Install the library
in your project environment, then run this from its root (requires [`uv`](https://docs.astral.sh/uv/)):

```shell
uvx library-skills install --skill http-client-usage --yes
```

For Claude Code, add `--claude`. The instructions are shipped with the library version.

## Usage

The library provides `HttpClient` and `AsyncHttpClient`. Use them directly or subclass them to create custom clients.

All options (logging, request identification, tracing, default headers, and others) work the same for both clients. Synchronous event hooks are automatically wrapped as async when used with `AsyncHttpClient`.

### Example:

#### Base client example

```python
import logging

from http_client import HttpClient, options
from httpx import BasicAuth

with HttpClient(
    "https://api.example.com",
    options.with_auth(BasicAuth("demo", "example-password")),
    options.with_logger(logging.getLogger(__name__)),
) as client:
    response = client.post("/test_method", json={"test": True})
    response.raise_for_status()
```

#### Custom client example

```python
import logging

from http_client import HttpClient, options
from httpx import Response


class CustomHttpClient(HttpClient):
    def test_method(self, data: dict[str, object]) -> Response:
        """Send data to a service method.

        :param data: JSON request body.
        :return: service response.
        """
        request = self.client.build_request("POST", "/test_method", json=data)
        return self.send(request)


with CustomHttpClient(
    "https://api.example.com",
    options.with_logger(logging.getLogger(__name__)),
    options.with_autogenerated_request_id(),
    options.with_autogenerated_traceparent(),
) as client:
    response = client.test_method({"test": True})
    response.raise_for_status()
```

#### Asynchronous client example

All standard options, including synchronous event hooks, work unchanged with `AsyncHttpClient`:

```python
import logging

from http_client import AsyncHttpClient, options

async with AsyncHttpClient(
    "https://api.example.com",
    options.with_logger(logging.getLogger(__name__)),
    options.with_autogenerated_request_id(),
) as client:
    response = await client.post("/test_method", json={"test": True})
    response.raise_for_status()
```

### Configuration
Configure the client with `ClientOption` options. Each option receives the actual client instance and runs in its constructor. Register hooks and decorators with `add_request_hook`, `add_response_hook`, `add_send_decorator`, and `add_stream_decorator`:

```python
from http_client import BaseHttpClient, ClientOption
from httpx import Request


def with_region(region: str) -> ClientOption:
    """Add a region to outgoing request headers.

    :param region: region identifier.
    :return: client option.
    """

    def add_region(request: Request) -> None:
        request.headers["X-Region"] = region

    def option(client: BaseHttpClient) -> None:
        client.add_request_hook(add_region, prepend=True)

    return option
```

Registration is available while constructor options run. After client creation, its transport is already configured.

## Available options

The `options` module provides options for configuring the HTTP client. Combine them when creating a client instance.

### EventHooks

Event hook options let you process requests and responses:

- `with_request_hook` — adds a function to process requests before sending
- `with_response_hook` — adds a function to process responses after receiving

These options provide a flexible way to handle each client request and/or response.

Usage example:

```python
import logging

from httpx import Request, Response

from http_client import HttpClient, options


def my_request_hook(request: Request) -> None:
    # process the request before sending it
    request.headers["X-Custom-Header"] = "Custom Value"


def my_response_hook(response: Response) -> None:
    # process the response after receiving it
    logging.getLogger(__name__).info("Response status code: %s", response.status_code)


client = HttpClient(
    "http://example.com",
    options.with_request_hook(my_request_hook),
    options.with_response_hook(my_response_hook),
)
```

With `AsyncHttpClient`, synchronous hooks are automatically wrapped as async. You can also pass native async hooks:

```python
from http_client import AsyncHttpClient, options
from httpx import Request


async def my_async_request_hook(request: Request) -> None:
    request.headers["X-Async"] = "true"


client = AsyncHttpClient(
    "http://example.com",
    options.with_request_hook(my_async_request_hook),
)
```

See the [`httpx` documentation](https://www.python-httpx.org/advanced/event-hooks/) for more about hooks.

### Decorating the send method

The HTTP client supports decorating its `send` method with `with_send_decorator`. This lets you add behavior before sending the HTTP request and after receiving the response.

`with_send_decorator` applies the decorator pattern to `send`, letting you:
- Preprocess the request
- Postprocess the response
- Implement retries, logging, caching, monitoring, and other features

Example of a request logging decorator:

```python
from functools import wraps
from http_client import HttpClient, options
from http_client.types import SendOptions, SyncSendMethod
from typing import Unpack
import logging
import httpx


def logging_decorator(func: SyncSendMethod) -> SyncSendMethod:
    @wraps(func)
    def wrapper(request: httpx.Request, **kwargs: Unpack[SendOptions]) -> httpx.Response:
        logging.getLogger(__name__).info("Sending request: %s %s", request.method, request.url)
        response = func(request, **kwargs)
        logging.getLogger(__name__).info("Received response: %s", response.status_code)
        return response

    return wrapper


client = HttpClient("https://example.com", options.with_send_decorator(logging_decorator))
```

For the asynchronous HTTP client, the decorator must be asynchronous:

```python
from functools import wraps
from http_client import AsyncHttpClient, options
from http_client.types import AsyncSendMethod, SendOptions
from typing import Unpack
import logging
import httpx


def async_logging_decorator(func: AsyncSendMethod) -> AsyncSendMethod:
    @wraps(func)
    async def wrapper(request: httpx.Request, **kwargs: Unpack[SendOptions]) -> httpx.Response:
        logging.getLogger(__name__).info("Sending request: %s %s", request.method, request.url)
        response = await func(request, **kwargs)
        logging.getLogger(__name__).info("Received response: %s", response.status_code)
        return response

    return wrapper


client = AsyncHttpClient("https://example.com", options.with_send_decorator(async_logging_decorator))
```

You can apply several decorators. Each new decorator wraps the previous one, so the last one added is outermost and runs first:

```python
client = HttpClient(
    "https://example.com",
    options.with_send_decorator(decorator1),  # applied first, becomes innermost
    options.with_send_decorator(decorator2),  # applied second, becomes outermost
)
```

Request calls run in this order: `decorator2` → `decorator1` → `send`. The response passes through them in reverse: `decorator1` finishes first, then `decorator2`.

### Streaming requests

The HTTP client supports streaming requests in two ways:

#### The `stream` method

`stream(method, url, **kwargs)` is fully compatible with the `httpx` API. It takes an HTTP method and URL, builds a request, and returns a context manager:

```python
from http_client import HttpClient, options

client = HttpClient(
    "https://example.com",
    options.with_auth(auth),
    options.with_logger(logger),
)

# Streaming download (httpx contract)
with client.stream("GET", "/large-file.zip") as response:
    response.raise_for_status()

    with open("file.zip", "wb") as f:
        for chunk in response.iter_bytes(chunk_size=8192):
            f.write(chunk)
```

Asynchronous client example:

```python
from http_client import AsyncHttpClient, options

client = AsyncHttpClient(
    "https://example.com",
    options.with_auth(auth),
    options.with_logger(logger),
)

async with client.stream("GET", "/stream-data") as response:
    response.raise_for_status()

    async for line in response.aiter_lines():
        process_line(line)
```

#### The `stream_request` method

`stream_request(request, **kwargs)` takes a ready-made `httpx.Request` and supports all client options, including decorators:

```python
request = client.client.build_request("GET", "/large-file.zip")

with client.stream_request(request) as response:
    for chunk in response.iter_bytes(chunk_size=8192):
        f.write(chunk)
```

#### Decorating the stream_request method

Like `send`, `stream_request` can be decorated with `with_stream_decorator`:

```python
from functools import wraps
from contextlib import contextmanager
from http_client import HttpClient, options
from http_client.types import StreamOptions, SyncStreamRequestMethod
from collections.abc import Iterator
from typing import Unpack
import httpx
import logging


def stream_logging_decorator(func: SyncStreamRequestMethod) -> SyncStreamRequestMethod:
    @wraps(func)
    @contextmanager
    def wrapper(request: httpx.Request, **kwargs: Unpack[StreamOptions]) -> Iterator[httpx.Response]:
        logging.getLogger(__name__).info("Starting stream: %s %s", request.method, request.url)
        with func(request, **kwargs) as response:
            logging.getLogger(__name__).info("Stream status: %s", response.status_code)
            yield response
        logging.getLogger(__name__).info("Stream finished")

    return wrapper


client = HttpClient("https://example.com", options.with_stream_decorator(stream_logging_decorator))
```

**Important:** When logging streaming requests, the response body appears as `<streamed content>` because it has not yet been read when the response hook runs.

### The request method

`request(method, url, **kwargs)` sends HTTP requests with any method:

```python
response = client.request("PATCH", "/api/resource", json={"field": "value"})
```

### Timeouts

`with_timeout` sets different types of HTTP request timeouts:

```python
from httpx import Timeout
from http_client import HttpClient, options

# Set different timeout types
timeout = Timeout(
    connect=5.0,  # time to establish a connection
    read=10.0,  # time to read the response
    write=5.0,  # time to write the request
    pool=1.0,  # time to wait for a free connection in the pool
)

client = HttpClient("https://example.com", options.with_timeout(timeout))
```

### Authorization

`with_auth` configures HTTP request authorization. All `httpx` authorization mechanisms are supported. You can also implement your own; see the [`httpx` documentation](https://www.python-httpx.org/advanced/authentication/#custom-authentication-schemes).

```python
from httpx import BasicAuth, DigestAuth
from http_client import HttpClient, options

# Basic authorization
client = HttpClient("https://example.com", options.with_auth(BasicAuth(username="user", password="password")))

# Digest authorization
client = HttpClient("https://example.com", options.with_auth(DigestAuth(username="user", password="password")))

# Custom Bearer token authorization
client = HttpClient(
    "https://example.com", options.with_auth(lambda r: r.headers.update({"Authorization": f"Bearer {token}"}))
)
```

### Request identification

`with_autogenerated_request_id` automatically adds a unique identifier to each request's `X-Request-ID` header:

```python
from http_client import HttpClient, options

# Automatically generate X-Request-ID
client = HttpClient("https://example.com", options.with_autogenerated_request_id())
```

This is useful for tracing requests across microservices and for debugging.

### Tracing

`with_autogenerated_traceparent` automatically adds a `traceparent` tracing header compatible with W3C Trace Context:

```python
from http_client import HttpClient, options

# Add a traceparent header to every request
client = HttpClient("https://example.com", options.with_autogenerated_traceparent())
```

This lets you track requests across multiple services using different monitoring systems.

### Logging

`with_logger` configures HTTP request and response logging. Set shared defaults with a `LoggingConfig` object:

```python
from http_client import configure_logging
from http_client.config import LoggingConfig

configure_logging(options=LoggingConfig(json_indent=2, use_line_breaks=True))
```

Pass client-specific settings to `with_logger`:

```python
import logging
from http_client import HttpClient, options

# Configure a custom logger
logger = logging.getLogger("http_client")

client = HttpClient(
    "https://example.com",
    options.with_logger(
        logger,
        sanitize_keys=["authorization", "password"],  # hide secret data
    ),
)
```

`sanitize_keys` hides sensitive information in logs, such as authorization tokens and passwords.

Common secrets are masked **by default**, even without `sanitize_keys`: `authorization`, `proxy-authorization`, `cookie`, `set-cookie`, `x-api-key`, `api-key`, `api_key`, `access_token`, `refresh_token`, `password`, `secret`, `token`. Keys from `sanitize_keys` extend this list.

Disable the default protection with an explicit flag, for example when debugging in a private environment:

```python
options.with_logger(logger, use_default_sanitize_keys=False)
```

⚠️ Masking applies only to access log records written by this library. `httpx` exceptions may contain the full URL, including query parameters, so do not pass secrets in URLs.

### Closing connections

The connection pool is created in the client constructor, so close the client. A context manager is the simplest way:

```python
with HttpClient("https://example.com") as client:
    client.get("/api/resource")

async with AsyncHttpClient("https://example.com") as client:
    await client.get("/api/resource")
```

For a long-lived client created once per application, use explicit methods:

```python
client.close()  # HttpClient
await client.aclose()  # AsyncHttpClient
```

A closed client cannot be reopened: like `httpx`, this raises `RuntimeError`.

### TLS connection

The library provides two options to configure TLS connections. Both work the same with `HttpClient` and `AsyncHttpClient`.

1. `with_verify_context` — sets the TLS verification context:

```python
import ssl
from http_client import HttpClient, options

# Create an SSL context
ssl_context = ssl.create_default_context()
ssl_context.load_verify_locations("/path/to/ca-certificates.crt")

# Set the SSL context
client = HttpClient("https://example.com", options.with_verify_context(verify=ssl_context))
```

2. `use_truststore` — uses the system certificate store for verification:

```python
from http_client import HttpClient, options

# Use the system certificate store
client = HttpClient("https://example.com", options.use_truststore())
```

To use `use_truststore`, install the extra dependency:
```
pip install "cloudru-qa-http-client[truststore]"
```

### Default headers

`with_default_headers` sets default headers sent with all client requests.

```python
from http_client import HttpClient, options

headers = {
    "X-My-Custom-Header": "Hello, world!",
    "X-Another-Header": "Another value",
}

client = HttpClient(
    "https://example.com",
    options.with_default_headers(headers),
)
```
## Development

The library is developed in the [qa-libraries](https://github.com/cloud-ru-tech/qa-libraries) monorepo
alongside the other `cloudru-qa-*` libraries. You need [`uv`](https://docs.astral.sh/uv/) 0.11.14
or later, GNU Make, and [`checkmake`](https://github.com/checkmake/checkmake). Run commands from the repository root:

```shell
make sync                     # install dependencies for all libraries from uv.lock
make test-http-client         # tests for this library
make typecheck-http-client    # mypy for this library
make check                    # formatting, ruff, mypy, and tests for the entire workspace
```

⚠️ **Keep** the library lightweight and general-purpose. Discuss new dependencies
only when you are sure all library users need them.

Tests run on Python 3.13 and 3.14. Create a separate environment for another version:
`make sync test-http-client PYTHON_VERSION=3.14`.

Demo scenarios with real requests to local [go-httpbin](https://github.com/mccutchen/go-httpbin)
are in `examples/httpbin`. Docker must be running; run commands from `packages/http-client`:

```shell
docker compose up -d
uv run --no-sync python -m examples.httpbin
docker compose down
```

Publish a release with the git tag `http-client/v<version>`. The tag version must match the version
in `packages/http-client/pyproject.toml`.
