hyperion-http 1.0.0

A production-ready, batteries-included HTTP client for Python

Hyperion HTTP

A production-ready, batteries-included HTTP client library for Python
Born from the best ideas of two HTTP client implementations — merged, fixed, and polished into one coherent framework.

---

Why Hyperion?

Feature	requests	Hyperion
Simple HTTP GET/POST	✅	✅
Connection pooling (keep-alive)	✅	✅
Authentication (Basic, Digest, Bearer, API-Key)	✅	✅
Cookie jar	✅	✅
Multipart file upload	✅	✅
Streaming responses	✅	✅
SSL/TLS + client certs	✅	✅
Response caching (memory + disk)	❌	✅
Circuit breaker pattern	❌	✅
Token-bucket rate limiting	❌	✅
Middleware chain	❌	✅
Per-endpoint request metrics	❌	✅
DNS result caching	❌	✅
Zero required dependencies	❌	✅

---

Installation

pip install hyperion-http

With optional extras for broader compression support:

pip install hyperion-http[compression]   # brotli + zstandard
pip install hyperion-http[full]          # everything

Or from source:

git clone https://github.com/dhaval-vedra/hyperion-http
cd hyperion-http
pip install -e .

---

Quick Start

import hyperion

# One-line requests — just like requests
r = hyperion.get("https://httpbin.org/get", params={"hello": "world"})
print(r.status_code)   # 200
print(r.json())        # dict

# POST JSON
r = hyperion.post(
    "https://api.example.com/items",
    json={"name": "Widget", "price": 9.99},
)
r.raise_for_status()

# Upload a file
with open("photo.jpg", "rb") as f:
    r = hyperion.post(
        "https://api.example.com/photos",
        files={"photo": ("photo.jpg", f, "image/jpeg")},
        data={"caption": "Sunset"},
    )

---

Sessions

Use a Session for multiple requests to the same host — it reuses connections and shares headers/cookies:

from hyperion import Session, HTTPBasicAuth

with Session(timeout=10.0) as s:
    s.headers["X-App-Version"] = "2.0"
    s.headers["Authorization"] = "Bearer my_token"

    profile = s.get("https://api.example.com/me")
    items   = s.get("https://api.example.com/items", params={"page": 1})
    created = s.post("https://api.example.com/items", json={"name": "New"})

---

Authentication

from hyperion import HTTPBasicAuth, HTTPDigestAuth, BearerTokenAuth, APIKeyAuth

# Basic
r = hyperion.get("https://api.example.com/", auth=HTTPBasicAuth("user", "pass"))

# Digest
r = hyperion.get("https://api.example.com/", auth=HTTPDigestAuth("user", "pass"))

# Bearer token / JWT
r = hyperion.get("https://api.example.com/", auth=BearerTokenAuth("eyJhbG..."))

# API key in header
r = hyperion.get("https://api.example.com/", auth=APIKeyAuth("sk-abc123", header_name="X-API-Key"))

# API key in query string
r = hyperion.get("https://api.example.com/", auth=APIKeyAuth("sk-abc123", in_query=True))

---

Response Caching

from hyperion import Session, CachePolicy

with Session(
    enable_cache=True,
    cache_policy=CachePolicy.MEMORY_AND_DISK,
    cache_ttl=300,
) as s:
    r1 = s.get("https://api.example.com/data", use_cache=True)  # network hit
    r2 = s.get("https://api.example.com/data", use_cache=True)  # from cache ⚡

---

Circuit Breaker

Prevent cascading failures when a downstream service is unhealthy:

from hyperion import Session
from hyperion.exceptions import CircuitBreakerError

with Session(
    enable_circuit_breaker=True,
    circuit_failure_threshold=5,
    circuit_recovery_timeout=60.0,
) as s:
    try:
        r = s.get("https://flaky-service.com/api")
    except CircuitBreakerError:
        print("Service is down — using fallback")

---

Rate Limiting

from hyperion import Session

# 20 requests per second
with Session(rate_limit=(20, 1.0)) as s:
    for url in urls:
        r = s.get(url)   # automatically throttled

---

Middleware

from hyperion import Session
from hyperion.middleware import LoggingMiddleware, TimingMiddleware, Middleware

# Built-in
with Session() as s:
    s.add_middleware(LoggingMiddleware())
    s.add_middleware(TimingMiddleware())
    r = s.get("https://api.example.com/")
    print(f"Took {r.elapsed:.3f}s")

# Custom middleware
class RetryOn503(Middleware):
    def process_response(self, response, request_data):
        if response.status_code == 503:
            print("503 received — you could retry here")
        return response

    def process_error(self, error, request_data):
        from hyperion.models import Response
        # Return a fallback response instead of propagating the error
        return Response(503, {}, b"Fallback", request_data["url"])

---

Request Metrics

with Session() as s:
    for _ in range(10):
        s.get("https://api.example.com/data")

    metrics = s.get_metrics()
    for host, stats in metrics.items():
        print(f"{host}: {stats['avg_response_time_ms']:.1f}ms avg, "
              f"{stats['success_rate']:.0f}% success")

---

Project Structure

hyperion-http/
├── hyperion/                    # Main package
│   ├── __init__.py              # Public API — import everything from here
│   ├── client.py                # Session + module-level functional API
│   ├── models.py                # Response, PreparedRequest, RequestMetrics
│   ├── auth.py                  # HTTPBasicAuth, HTTPDigestAuth, BearerTokenAuth, APIKeyAuth
│   ├── cookies.py               # Cookie, CookieJar
│   ├── adapters.py              # HTTPAdapter, ConnectionPool, PooledConnection
│   ├── cache.py                 # AdvancedCache, CacheEntry, CachePolicy
│   ├── circuit_breaker.py       # CircuitBreaker, CircuitBreakerState
│   ├── rate_limiter.py          # RateLimiter (token bucket)
│   ├── dns_cache.py             # DNSCache
│   ├── middleware.py            # Middleware base + LoggingMiddleware, TimingMiddleware…
│   ├── hooks.py                 # EventHooks
│   ├── encoders.py              # MultipartEncoder, URLEncodedEncoder, StreamingBody
│   ├── exceptions.py            # All custom exceptions
│   ├── utils.py                 # URL/header/compression/backoff helpers
│   └── py.typed                 # PEP 561 marker
│
├── tests/                       # Unit tests (pytest)
│   ├── test_auth.py
│   ├── test_cache.py
│   ├── test_circuit_breaker.py
│   ├── test_cookies.py
│   ├── test_exceptions.py
│   ├── test_models.py
│   ├── test_rate_limiter.py
│   └── test_utils.py
│
├── examples/
│   ├── basic_usage.py           # GET, POST, auth, upload, streaming, cookies
│   ├── advanced_features.py     # Caching, circuit breaker, rate limiter, metrics
│   └── middleware_example.py    # Built-in and custom middleware
│
├── docs/
│   ├── quickstart.md            # 5-minute getting-started guide
│   ├── advanced.md              # Deep-dive into advanced features
│   └── api_reference.md         # Complete class/method reference
│
├── README.md
├── CHANGELOG.md
├── LICENSE                      # MIT
├── pyproject.toml               # Build config, extras, tool settings
├── requirements.txt             # Runtime deps (none required)
└── requirements-dev.txt         # Dev/test deps

---

Bugs Fixed (vs. original sources)

This library merges and fixes two original prototype files:

Bug	Source	Fix
AdvancedCache.get accessed row[4] but only 4 columns were SELECTed (index 0–3) — IndexError	v2	Fixed query to SELECT the expires_at column correctly
RequestDeduplicator used asyncio.Future in a synchronous threading context — RuntimeError	v2	Replaced with threading.Event + shared result dict
CircuitBreakerError was used before it was defined	v2	Moved all exceptions to exceptions.py (imported at top)
HTTPError was redefined at the bottom of the file, shadowing the first definition	v2	Deduplicated into exceptions.py
_request_http1, _request_http2, _parse_response were pass stubs — TypeError: 'NoneType' is not subscriptable	v2	Replaced with a complete working HTTP/1.1 socket implementation
urllib3_timeout imported but never used	v1	Removed unused import
AuthBase.__call__ signature declared request: bytes but subclasses received headers: Dict — TypeError	v1	Aligned signature to headers: Dict[str, str]
isinstance(data, Iterator) — Iterator is a generic alias, not valid for isinstance in Python 3.9+	v1	Replaced with collections.abc.Iterator
iter_content bare except: pass silently swallowed decode errors	v1	Replaced with explicit errors="replace"

---

Running the Tests

# Install dev dependencies
pip install -r requirements-dev.txt

# Run all tests
pytest

# With coverage
pytest --cov=hyperion --cov-report=term-missing

---

Running Examples

# Basic usage
python examples/basic_usage.py

# Advanced features
python examples/advanced_features.py

# Middleware
python examples/middleware_example.py

---

Documentation

Document	Description
Quickstart	Get running in 5 minutes
Advanced Features	Caching, circuit breaker, rate limiting, middleware, metrics
API Reference	Complete class and method reference
Changelog	What changed in each version

---

License

MIT — free to use, modify, and distribute.