tessl/pypi-urllib3

HTTP library with thread-safe connection pooling, file post support, user friendly interface, and more.

—

Pending

Quality

Pending

Does it follow best practices?

Impact

Pending

No eval scenarios have been run

Overview

Eval results

Files

Response Handling

Name: tessl/pypi-urllib3
Author: tessl

Comprehensive HTTP response objects with content decoding, streaming support, metadata access, and flexible data consumption patterns. urllib3 provides rich response handling capabilities for various use cases.

Capabilities

BaseHTTPResponse

Base interface for all HTTP response objects, providing core functionality for accessing response data, headers, and metadata.

class BaseHTTPResponse:
    @property
    def status(self) -> int:
        """HTTP status code (200, 404, 500, etc.)"""

    @property
    def headers(self) -> HTTPHeaderDict:
        """Response headers as case-insensitive dictionary"""

    @property
    def data(self) -> bytes:
        """Complete response body as bytes"""

    def read(self, amt=None, decode_content=None) -> bytes:
        """
        Read response data.

        Parameters:
        - amt: Maximum bytes to read (None for all)
        - decode_content: Whether to decode compressed content

        Returns:
        bytes: Response data
        """

    def stream(self, amt=None, decode_content=None):
        """
        Stream response data in chunks.

        Parameters:
        - amt: Chunk size in bytes
        - decode_content: Whether to decode compressed content

        Yields:
        bytes: Chunks of response data
        """

    def close(self):
        """Close the response and release connection back to pool"""

    def isclosed(self) -> bool:
        """Check if response has been closed"""

HTTPResponse

Full HTTP response implementation with advanced features including automatic content decoding, JSON parsing, and comprehensive metadata access.

class HTTPResponse(BaseHTTPResponse):
    @property
    def reason(self) -> str:
        """HTTP reason phrase (OK, Not Found, etc.)"""

    @property
    def version(self) -> int:
        """HTTP version (11 for HTTP/1.1, 20 for HTTP/2)"""

    @property
    def connection(self):
        """Underlying connection object"""

    @property
    def retries(self) -> Retry:
        """Retry configuration that was used"""
    
    @property
    def url(self) -> str:
        """Final URL after redirects"""

    def json(self) -> any:
        """
        Parse response body as JSON.

        Returns:
        any: Parsed JSON data

        Raises:
        json.JSONDecodeError: If response is not valid JSON
        """

    def getheader(self, name: str, default=None) -> str:
        """
        Get single header value.

        Parameters:
        - name: Header name (case-insensitive)
        - default: Default value if header not found

        Returns:
        str: Header value or default
        """

    def getheaders(self, name: str) -> list[str]:
        """
        Get all values for a header (for multi-value headers).

        Parameters:
        - name: Header name (case-insensitive)

        Returns:
        list[str]: List of header values
        """

HTTPHeaderDict

Case-insensitive dictionary for HTTP headers with support for multi-value headers.

class HTTPHeaderDict:
    def __init__(self, headers=None):
        """
        Case-insensitive HTTP header dictionary.

        Parameters:
        - headers: Initial headers (dict, list of tuples, or HTTPHeaderDict)
        """

    def add(self, key: str, val: str):
        """Add header value (supports multiple values per header)"""

    def extend(self, other):
        """Extend headers from another dict/HTTPHeaderDict"""

    def get(self, key: str, default=None) -> str:
        """Get header value (case-insensitive)"""

    def get_all(self, key: str) -> list[str]:
        """Get all values for header"""

    def items(self):
        """Iterate over (name, value) pairs"""

    def values(self):
        """Iterate over header values"""

    def keys(self):
        """Iterate over header names"""

Usage Examples

Basic Response Handling

import urllib3

http = urllib3.PoolManager()
resp = http.request('GET', 'https://httpbin.org/json')

# Access response properties
print(f"Status: {resp.status}")           # 200
print(f"Reason: {resp.reason}")           # OK
print(f"Headers: {resp.headers}")         # HTTPHeaderDict
print(f"Content-Type: {resp.headers['content-type']}")
print(f"Data: {resp.data.decode('utf-8')}")  # Response body as string

# Close response to return connection to pool
resp.close()

JSON Response Handling

import urllib3

http = urllib3.PoolManager()
resp = http.request('GET', 'https://jsonplaceholder.typicode.com/posts/1')

# Parse JSON response
data = resp.json()
print(f"Title: {data['title']}")
print(f"Body: {data['body']}")

resp.close()

Streaming Large Responses

import urllib3

http = urllib3.PoolManager()
resp = http.request('GET', 'https://httpbin.org/stream/1000', 
                   preload_content=False)  # Don't load entire response

# Stream response in chunks
total_bytes = 0
for chunk in resp.stream(1024):  # 1KB chunks
    total_bytes += len(chunk)
    print(f"Received {len(chunk)} bytes, total: {total_bytes}")

print(f"Total downloaded: {total_bytes} bytes")
resp.close()

Reading Response in Chunks

import urllib3

http = urllib3.PoolManager()
resp = http.request('GET', 'https://httpbin.org/bytes/10000',
                   preload_content=False)

# Read specific amounts
chunk1 = resp.read(1024)  # Read first 1KB
chunk2 = resp.read(2048)  # Read next 2KB
remaining = resp.read()   # Read rest

print(f"Chunk 1: {len(chunk1)} bytes")
print(f"Chunk 2: {len(chunk2)} bytes") 
print(f"Remaining: {len(remaining)} bytes")

resp.close()

Header Manipulation

import urllib3

http = urllib3.PoolManager()
resp = http.request('GET', 'https://httpbin.org/response-headers',
                   fields={'Content-Type': 'application/json',
                          'Custom-Header': 'MyValue'})

# Access headers in various ways
print(f"Content-Type: {resp.headers['content-type']}")
print(f"Content-Type: {resp.getheader('content-type')}")

# Check if header exists
if 'custom-header' in resp.headers:
    print(f"Custom header: {resp.headers['custom-header']}")

# Iterate over all headers
for name, value in resp.headers.items():
    print(f"{name}: {value}")

resp.close()

Working with Multi-Value Headers

import urllib3

# Create HTTPHeaderDict
headers = urllib3.HTTPHeaderDict()
headers.add('Set-Cookie', 'session=abc123; Path=/')
headers.add('Set-Cookie', 'theme=dark; Path=/')

# Get all values for a header
cookies = headers.get_all('set-cookie')
print(f"All cookies: {cookies}")

# Get first value only
first_cookie = headers.get('set-cookie')
print(f"First cookie: {first_cookie}")

Response Context Management

import urllib3

http = urllib3.PoolManager()

# Using context manager (recommended)
with http.request('GET', 'https://httpbin.org/json') as resp:
    data = resp.json()
    print(f"Status: {resp.status}")
    print(f"Data: {data}")
    # Response is automatically closed

# Manual management
resp = http.request('GET', 'https://httpbin.org/json')
try:
    data = resp.json()
    # Process data
finally:
    resp.close()  # Always close

Handling Different Content Types

import urllib3

http = urllib3.PoolManager()

# JSON response
json_resp = http.request('GET', 'https://httpbin.org/json')
json_data = json_resp.json()

# Text response
text_resp = http.request('GET', 'https://httpbin.org/html')
html_content = text_resp.data.decode('utf-8')

# Binary response
binary_resp = http.request('GET', 'https://httpbin.org/bytes/1000')
binary_data = binary_resp.data  # Raw bytes

# Close all responses
json_resp.close()
text_resp.close()
binary_resp.close()

Error Response Handling

import urllib3

http = urllib3.PoolManager()

resp = http.request('GET', 'https://httpbin.org/status/404')

if resp.status == 404:
    print("Resource not found")
elif resp.status >= 400:
    print(f"Client error: {resp.status} {resp.reason}")
elif resp.status >= 500:
    print(f"Server error: {resp.status} {resp.reason}")
else:
    print("Success!")

# Response data might contain error details
error_body = resp.data.decode('utf-8')
print(f"Error response: {error_body}")

resp.close()