Skip to main content

Library for HTTP operations.

Project description

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)

Installation

Install the package from PyPI using pip:

pip install universal-common-net-http

Usage

Basic requests

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

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

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

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

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

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

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

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

Project details


Download files

Download the file for your platform. If you're not sure which to choose, learn more about installing packages.

Source Distribution

universal_common_net_http-1.0.1.tar.gz (18.2 kB view details)

Uploaded Source

Built Distribution

If you're not sure about the file name format, learn more about wheel file names.

universal_common_net_http-1.0.1-py3-none-any.whl (17.2 kB view details)

Uploaded Python 3

File details

Details for the file universal_common_net_http-1.0.1.tar.gz.

File metadata

File hashes

Hashes for universal_common_net_http-1.0.1.tar.gz
Algorithm Hash digest
SHA256 bed11d36a0e4a2b64257fcbbecd0f37646341fe7a102f620bdee46b8684f9002
MD5 29d2c1a919a4f76a34ed05b0f11acde1
BLAKE2b-256 7bc262ab2e61f6240c5911077b8312b02eb9a6fbcace29759a1b6e34fc3bfa4a

See more details on using hashes here.

File details

Details for the file universal_common_net_http-1.0.1-py3-none-any.whl.

File metadata

File hashes

Hashes for universal_common_net_http-1.0.1-py3-none-any.whl
Algorithm Hash digest
SHA256 6c7985e2f01adae471cf7bd251b324481a8f3a52add779d46261982e05fc846b
MD5 09c8b6b26ed0d5f384d0f73425a403de
BLAKE2b-256 c2f58c380838a4db9ed8366c1b55da908ee84fac747cf24a7a9eb7a102f2eef6

See more details on using hashes here.

Supported by

AWS Cloud computing and Security Sponsor Datadog Monitoring Depot Continuous Integration Fastly CDN Google Download Analytics Pingdom Monitoring Sentry Error logging StatusPage Status page