Metadata-Version: 2.4
Name: universal-common-net-http
Version: 1.0.1
Summary: Library for HTTP operations.
Author-email: Andrew Ong <ong.andrew@gmail.com>
License-File: LICENSE
Classifier: Development Status :: 5 - Production/Stable
Classifier: License :: CC0 1.0 Universal (CC0 1.0) Public Domain Dedication
Classifier: Operating System :: OS Independent
Classifier: Programming Language :: Python :: 3
Requires-Python: >=3.9
Provides-Extra: test
Requires-Dist: pytest; extra == 'test'
Requires-Dist: pytest-asyncio; extra == 'test'
Description-Content-Type: text/markdown

# universal-common-net-http

A Python library inspired by .NET's `System.Net.Http`, providing an `HttpClient`/`HttpRequestMessage`/`HttpResponseMessage`-shaped HTTP stack plus a higher-level `HttpServiceClient` pipeline for building typed API clients. Built entirely on the Python standard library - no third-party runtime dependencies.

## Features

- `HttpClient` with `GetAsync`/`PostAsync`/`PutAsync`/`DeleteAsync`/`SendAsync`-style methods, base address, default headers, and configurable timeout
- Typed request content: `StringContent`, `JsonContent`, `ByteArrayContent`, `FormUrlEncodedContent`, `StreamContent`
- Case-insensitive, multi-value headers (`HttpHeaders`), matching real HTTP semantics
- Automatic redirect following (301/302/303/307/308) with safe defaults: refuses to downgrade HTTPS to plaintext, strips `Authorization`/`Cookie` on cross-host hops
- Automatic cookie handling via the standard library's `http.cookiejar`
- Automatic gzip/deflate response decompression
- Connection pooling and keep-alive reuse, with idle-timeout and lifetime eviction
- True streaming: upload from a file-like object/iterable without buffering it all in memory, and download via `get_stream_async` without buffering the response body
- `HttpServiceClient`: a subclassable pipeline with pre/post-request hooks and success/failure handlers, for building typed API clients
- Client certificates, proxy support (including HTTPS tunneling), preemptive Basic auth, and a custom server-certificate validation hook (for pinning or accepting self-signed certs in dev/test)
- `DelegatingHandler` for composing cross-cutting behavior (logging, retries, metrics) around the transport
- `cancel_pending_requests()` to abort in-flight requests
- HTTP/1.0 and HTTP/1.1 support, with a clear `NotImplementedError` for HTTP/2 and HTTP/3 rather than a silent downgrade or hang (see [Design notes](#design-notes))

## Installation

Install the package from PyPI using pip:

```bash
pip install universal-common-net-http
```

## Usage

### Basic requests

```python
import asyncio
from universal_common_net_http import HttpClient, JsonContent

async def main():
    async with HttpClient() as client:
        response = await client.get_async("https://api.example.com/users/1")
        response.ensure_success_status_code()
        user = await response.content.read_as_json_async()

        response = await client.post_async(
            "https://api.example.com/users",
            JsonContent({"name": "Ada Lovelace"}),
        )
        response.ensure_success_status_code()

asyncio.run(main())
```

### Typed content

```python
from universal_common_net_http import StringContent, JsonContent, FormUrlEncodedContent, ByteArrayContent

await client.post_async(url, JsonContent({"key": "value"}))
await client.post_async(url, StringContent("plain text body"))
await client.post_async(url, FormUrlEncodedContent({"field": "value"}))
await client.post_async(url, ByteArrayContent(b"raw bytes", media_type="application/octet-stream"))
```

### Streaming

```python
from universal_common_net_http import StreamContent

# Download without buffering the whole response body in memory.
async with HttpClient() as client:
    async for chunk in client.get_stream_async("https://example.com/large-file.zip"):
        handle(chunk)

# Upload without buffering the whole request body in memory. Omit `length` to
# send Transfer-Encoding: chunked instead (requires HTTP/1.1).
with open("large-file.zip", "rb") as file:
    content = StreamContent(file, media_type="application/zip", length=file_size)
    await client.post_async("https://example.com/upload", content)
```

### `HttpServiceClient`: building a typed API client

```python
from universal_common_net_http import HttpServiceClient, HttpRequestMessage, HttpResponseMessage

class MyApiClient(HttpServiceClient):
    def create_http_client(self):
        client = super().create_http_client()
        client.set_base_address("https://api.example.com")
        client.default_request_headers_add("Accept", "application/json")
        return client

    async def pre_process_http_request_message_async(self, request: HttpRequestMessage):
        request.headers_add("Authorization", "Bearer <token>")

    async def handle_non_success_status_code_async(self, response: HttpResponseMessage):
        response.ensure_success_status_code()

async def usage():
    async with MyApiClient() as client:
        response = await client.get_async("/users/1")
        return await response.content.read_as_json_async()
```

### Redirects and cookies

```python
from universal_common_net_http import HttpClient, HttpClientHandler

handler = HttpClientHandler()
handler.allow_auto_redirect = True   # default
handler.max_automatic_redirects = 50 # default
handler.use_cookies = True           # default; cookies persist on handler.cookies across requests

async with HttpClient(handler) as client:
    ...
```

### Timeouts and cancellation

```python
async with HttpClient() as client:
    client.timeout = 10  # seconds; set to None for no timeout

    # Aborts every request currently in flight on this client.
    client.cancel_pending_requests()
```

### Composable handlers

```python
from universal_common_net_http import DelegatingHandler, HttpClientHandler, HttpCompletionOption

class LoggingHandler(DelegatingHandler):
    async def send_async(self, request, timeout=None, completion_option=HttpCompletionOption.RESPONSE_CONTENT_READ):
        print(f"-> {request.method} {request.request_uri}")
        response = await super().send_async(request, timeout=timeout, completion_option=completion_option)
        print(f"<- {response.status_code}")
        return response

async with HttpClient(LoggingHandler(HttpClientHandler())) as client:
    ...
```

### Certificates and proxies

```python
handler = HttpClientHandler()
handler.client_certificates = [("cert.pem", "key.pem")]
handler.credentials = ("username", "password")  # sent preemptively as HTTP Basic auth
handler.use_proxy = True
handler.proxy = "http://myproxy:8080"

# Custom certificate validation (e.g. pinning, or accepting a self-signed cert in dev/test).
# When set, this callback is the sole authority on whether to keep the connection.
handler.server_certificate_custom_validation = lambda der_cert, sock: True
```

## Design notes

This library is intentionally built on the Python standard library alone - `http.client`, `ssl`, `asyncio`, `http.cookiejar` - with no runtime dependencies. That keeps the install footprint minimal, but it does mean a few things are out of scope on purpose rather than by oversight:

- **HTTP/2 and HTTP/3 are not supported.** Python's standard library has no HTTP/2 client (no binary framing, HPACK, or stream multiplexing) and no HTTP/3/QUIC support. Setting `HttpRequestMessage.version` to `"2.0"` or `"3.0"` raises `NotImplementedError` rather than silently downgrading to HTTP/1.1 or hanging.
- **Brotli response decompression is not supported** (no `br` in the default `Accept-Encoding`); gzip and deflate are.
- **Only Basic auth is automatic**, and it's sent preemptively rather than only after a challenge. Digest, NTLM, and Negotiate are not implemented.

## License

CC0 1.0 Universal (Public Domain Dedication).
