tessl/pypi-hishel

Persistent cache implementation for httpx and httpcore following RFC 9111 specification

1.48x

Overview

Eval results

Files

Cache Transports

Name: tessl/pypi-hishel
Rating: 0.74 (1 reviews)
Author: tessl

HTTPX transport implementations that add a caching layer on top of existing transports. These transports handle cache lookups, storage, validation, and seamlessly integrate with HTTPX's transport architecture.

Capabilities

Synchronous Cache Transport

HTTP transport that wraps existing synchronous transports to add caching capabilities.

class CacheTransport(httpx.BaseTransport):
    def __init__(self, *, transport: httpx.BaseTransport, storage=None, controller=None):
        """
        Synchronous HTTP transport with caching.
        
        Parameters:
        - transport: Underlying HTTPX transport to wrap
        - storage: Storage backend for cached responses (defaults to FileStorage)
        - controller: Cache controller for caching logic (defaults to Controller())
        """
    
    def handle_request(self, request: httpx.Request) -> httpx.Response:
        """Handle HTTP request with caching logic"""
    
    def close(self) -> None:
        """Close transport and storage resources"""
    
    def __enter__(self) -> "CacheTransport": ...
    def __exit__(self, *args) -> None: ...

Usage Examples:

import httpx
import hishel

# Wrap existing transport with caching
base_transport = httpx.HTTPTransport()
cache_transport = hishel.CacheTransport(
    transport=base_transport,
    storage=hishel.FileStorage(),
    controller=hishel.Controller()
)

# Use directly with httpx
with httpx.Client(transport=cache_transport) as client:
    response = client.get("https://api.example.com/data")

# Custom transport configuration
import ssl
context = ssl.create_default_context()
base_transport = httpx.HTTPTransport(
    verify=context,
    cert=('client.crt', 'client.key'),
    timeout=30.0
)
cache_transport = hishel.CacheTransport(transport=base_transport)

with httpx.Client(transport=cache_transport) as client:
    response = client.get("https://api.example.com/data")

Asynchronous Cache Transport

HTTP transport that wraps existing asynchronous transports to add caching capabilities.

class AsyncCacheTransport(httpx.AsyncBaseTransport):
    def __init__(self, *, transport: httpx.AsyncBaseTransport, storage=None, controller=None):
        """
        Asynchronous HTTP transport with caching.
        
        Parameters:
        - transport: Underlying async HTTPX transport to wrap
        - storage: Async storage backend for cached responses (defaults to AsyncFileStorage)
        - controller: Cache controller for caching logic (defaults to Controller())
        """
    
    async def handle_async_request(self, request: httpx.Request) -> httpx.Response:
        """Handle HTTP request with caching logic asynchronously"""
    
    async def aclose(self) -> None:
        """Close transport and storage resources asynchronously"""
    
    async def __aenter__(self) -> "AsyncCacheTransport": ...
    async def __aexit__(self, *args) -> None: ...

Usage Examples:

import httpx
import hishel
import asyncio

async def main():
    # Wrap existing async transport with caching
    base_transport = httpx.AsyncHTTPTransport()
    cache_transport = hishel.AsyncCacheTransport(
        transport=base_transport,
        storage=hishel.AsyncRedisStorage(),
        controller=hishel.Controller(allow_heuristics=True)
    )

    # Use directly with async httpx
    async with httpx.AsyncClient(transport=cache_transport) as client:
        response = await client.get("https://api.example.com/data")

    # Custom async transport configuration
    import ssl
    context = ssl.create_default_context()
    base_transport = httpx.AsyncHTTPTransport(
        verify=context,
        http2=True,
        limits=httpx.Limits(max_connections=100)
    )
    cache_transport = hishel.AsyncCacheTransport(transport=base_transport)

    async with httpx.AsyncClient(transport=cache_transport) as client:
        response = await client.get("https://api.example.com/data")

asyncio.run(main())

Cache Streams

Stream implementations for serving cached response content efficiently.

class CacheStream(httpx.SyncByteStream):
    def __init__(self, httpcore_stream: Iterable[bytes]):
        """
        Synchronous byte stream for cached responses.
        
        Parameters:
        - httpcore_stream: Iterable of bytes for response content
        """
    
    def __iter__(self) -> Iterator[bytes]:
        """Iterate over response content bytes"""
    
    def close(self) -> None:
        """Close the stream"""

class AsyncCacheStream(httpx.AsyncByteStream):
    def __init__(self, httpcore_stream: AsyncIterable[bytes]):
        """
        Asynchronous byte stream for cached responses.
        
        Parameters:
        - httpcore_stream: Async iterable of bytes for response content
        """
    
    async def __aiter__(self) -> AsyncIterator[bytes]:
        """Async iterate over response content bytes"""
    
    async def aclose(self) -> None:
        """Close the stream asynchronously"""

Transport Integration

Cache transports seamlessly integrate with HTTPX's transport system:

With Connection Pooling

import httpx
import hishel

# HTTP/1.1 with connection pooling
base_transport = httpx.HTTPTransport(
    limits=httpx.Limits(
        max_keepalive_connections=20,
        max_connections=100,
        keepalive_expiry=30.0
    )
)
cache_transport = hishel.CacheTransport(transport=base_transport)

with httpx.Client(transport=cache_transport) as client:
    response = client.get("https://api.example.com/data")

With HTTP/2 Support

import httpx
import hishel

# HTTP/2 with caching
base_transport = httpx.AsyncHTTPTransport(http2=True)
cache_transport = hishel.AsyncCacheTransport(transport=base_transport)

async with httpx.AsyncClient(transport=cache_transport) as client:
    response = await client.get("https://api.example.com/data")

With Custom Authentication

import httpx
import hishel

# Custom auth with caching
auth = httpx.DigestAuth("username", "password")
base_transport = httpx.HTTPTransport()
cache_transport = hishel.CacheTransport(transport=base_transport)

with httpx.Client(transport=cache_transport, auth=auth) as client:
    response = client.get("https://api.example.com/protected")

With Proxy Support

import httpx
import hishel

# Proxy transport with caching
proxy_transport = httpx.HTTPTransport(proxy="http://proxy.example.com:8080")
cache_transport = hishel.CacheTransport(transport=proxy_transport)

with httpx.Client(transport=cache_transport) as client:
    response = client.get("https://api.example.com/data")

Cache Transport Behavior

Request Flow

Cache Lookup: Check if request has a valid cached response
Freshness Check: Validate cached response freshness using Controller
Conditional Request: Send validation request if needed (ETag, Last-Modified)
Response Handling: Store new responses or update cached responses
Response Serving: Return cached or fresh response to client

Error Handling

Cache transports gracefully handle errors:

Storage Errors: Fall back to network requests if cache storage fails
Network Errors: Serve stale responses when allowed and available
Serialization Errors: Skip caching and serve fresh responses
Validation Errors: Treat as cache miss and fetch fresh response

504 Gateway Timeout

When network fails and no cached response is available:

# Returns 504 status code when network is unavailable
# and no cached response exists
response = client.get("https://unreachable.example.com/data")
assert response.status_code == 504

Advanced Usage

Custom Transport Chains

import httpx
import hishel

# Chain multiple transports
retry_transport = httpx.HTTPTransport(retries=3)
cache_transport = hishel.CacheTransport(transport=retry_transport)

# Custom middleware transport
class LoggingTransport(httpx.BaseTransport):
    def __init__(self, transport):
        self.transport = transport
    
    def handle_request(self, request):
        print(f"Requesting: {request.url}")
        response = self.transport.handle_request(request)
        print(f"Response: {response.status_code}")
        return response

logging_transport = LoggingTransport(cache_transport)

with httpx.Client(transport=logging_transport) as client:
    response = client.get("https://api.example.com/data")

Manual Transport Usage

import httpx
import hishel

# Use transport directly without client
transport = hishel.CacheTransport(
    transport=httpx.HTTPTransport(),
    storage=hishel.InMemoryStorage(),
    controller=hishel.Controller()
)

request = httpx.Request("GET", "https://api.example.com/data")
response = transport.handle_request(request)

transport.close()

Install with Tessl CLI

npx tessl i tessl/pypi-hishel

tessl/pypi-hishel