Response Handling¶
This guide covers how to work with HTTP responses in httpr.
The Response Object¶
Every request returns a Response object with all the information about the server's response:
import httpr
response = httpr.get("https://httpbin.org/get")
# Access response data
print(response.status_code) # HTTP status code
print(response.text) # Body as string
print(response.content) # Body as bytes
print(response.headers) # Response headers
print(response.cookies) # Response cookies
print(response.url) # Final URL (after redirects)
print(response.encoding) # Character encoding
Status Codes¶
Check the HTTP status code to determine if the request succeeded:
import httpr
response = httpr.get("https://httpbin.org/status/200")
print(response.status_code) # 200
response = httpr.get("https://httpbin.org/status/404")
print(response.status_code) # 404
print(response.reason_phrase) # "Not Found"
Instead of comparing integers by hand, use the status-class helpers:
import httpr
response = httpr.get("https://httpbin.org/status/404")
if response.is_success: # 2xx
print("Success!")
elif response.is_redirect: # 3xx
print("Redirect")
elif response.is_client_error: # 4xx
print("Client error")
elif response.is_server_error: # 5xx
print("Server error")
# `is_error` is True for any 4xx or 5xx response
print(response.is_error) # True
Available helpers: is_informational (1xx), is_success (2xx),
is_redirect (3xx), is_client_error (4xx), is_server_error (5xx),
is_error (4xx/5xx), and has_redirect_location (3xx with a Location
header).
Response Body¶
Text Content¶
Get the response body as a decoded string:
import httpr
response = httpr.get("https://httpbin.org/html")
print(response.text) # HTML content as string
The encoding is automatically detected from:
- The
Content-Typeheader charset - HTML meta charset tag (first 2048 bytes)
- Falls back to UTF-8
Access the detected encoding:
import httpr
response = httpr.get("https://httpbin.org/encoding/utf8")
print(response.encoding) # "utf-8"
Binary Content¶
Get the raw response body as bytes:
import httpr
response = httpr.get("https://httpbin.org/bytes/100")
print(type(response.content)) # <class 'bytes'>
print(len(response.content)) # 100
# Useful for binary files
response = httpr.get("https://httpbin.org/image/png")
with open("image.png", "wb") as f:
f.write(response.content)
JSON Content¶
Parse the response body as JSON:
import httpr
response = httpr.get("https://httpbin.org/json")
data = response.json()
print(type(data)) # <class 'dict'>
print(data["slideshow"]["title"])
Note
json() is a method, not a property. Call it with parentheses.
CBOR Content (Transparent Deserialization)¶
When the server returns Content-Type: application/cbor, the json() method automatically deserializes CBOR data:
import httpr
response = httpr.get("https://api.example.com/data")
# json() automatically detects and deserializes CBOR based on Content-Type
data = response.json() # Works transparently with CBOR!
print(type(data)) # <class 'dict'> or <class 'list'>
print(data)
You can also explicitly use cbor() if you know the data is CBOR:
import httpr
response = httpr.get("https://api.example.com/cbor-data")
data = response.cbor() # Explicitly deserialize as CBOR
CBOR is a binary serialization format that's more compact than JSON, making it ideal for:
- Large datasets: Smaller payload sizes compared to JSON
- IoT applications: Efficient data transfer for resource-constrained devices
- High-performance APIs: Faster serialization/deserialization
Transparent Usage
In most cases, you don't need to think about CBOR vs JSON. Just use response.json() and httpr will automatically handle the deserialization based on the Content-Type header.
HTML Conversion¶
httpr provides built-in HTML-to-text conversion using Rust's html2text crate:
import httpr
response = httpr.get("https://example.com")
# Convert HTML to Markdown
markdown = response.text_markdown
print(markdown)
# Convert HTML to plain text (no formatting)
plain = response.text_plain
print(plain)
# Convert HTML to rich text
rich = response.text_rich
print(rich)
This is useful for:
- Extracting readable content from web pages
- Processing HTML for text analysis
- Creating plain-text versions of HTML emails
Response Headers¶
Access response headers through the headers attribute:
import httpr
response = httpr.get("https://httpbin.org/response-headers?X-Custom=test")
# Headers are case-insensitive
print(response.headers["content-type"])
print(response.headers["Content-Type"]) # Same result
print(response.headers["CONTENT-TYPE"]) # Also works
# Check if header exists
if "x-custom" in response.headers:
print(response.headers["x-custom"])
# Get with default value
value = response.headers.get("x-missing", "default")
# Iterate over headers
for key, value in response.headers.items():
print(f"{key}: {value}")
# Get all header names
print(list(response.headers.keys()))
# Get all header values
print(list(response.headers.values()))
Common Headers¶
import httpr
response = httpr.get("https://httpbin.org/get")
# Content information
content_type = response.headers.get("content-type")
content_length = response.headers.get("content-length")
# Caching
cache_control = response.headers.get("cache-control")
etag = response.headers.get("etag")
# Server information
server = response.headers.get("server")
date = response.headers.get("date")
Response Cookies¶
Access cookies set by the server:
import httpr
response = httpr.get("https://httpbin.org/cookies/set?session=abc123")
# Cookies as a dictionary
print(response.cookies) # {"session": "abc123"}
# Access specific cookie
if "session" in response.cookies:
print(response.cookies["session"])
Cookie Persistence¶
With cookie_store=True (default), cookies are automatically stored and sent with subsequent requests:
import httpr
client = httpr.Client(cookie_store=True) # Default
# First request sets a cookie
client.get("https://httpbin.org/cookies/set?token=xyz")
# Cookie is automatically included in next request
response = client.get("https://httpbin.org/cookies")
print(response.json()["cookies"]) # {"token": "xyz"}
See the Cookie Handling guide for more details.
Final URL¶
After redirects, check the final URL:
import httpr
response = httpr.get("https://httpbin.org/redirect/3")
# The URL after following all redirects
print(response.url) # https://httpbin.org/get
This is useful for:
- Detecting redirects
- Getting the canonical URL
- Debugging redirect chains
Complete Example¶
Here's a comprehensive example of response handling:
import httpr
def fetch_and_process(url: str) -> dict:
"""Fetch a URL and return processed response data."""
response = httpr.get(url, timeout=10)
result = {
"url": response.url,
"status": response.status_code,
"success": 200 <= response.status_code < 300,
}
# Get content type
content_type = response.headers.get("content-type", "")
result["content_type"] = content_type
# Process based on content type
if "application/json" in content_type:
result["data"] = response.json()
elif "text/html" in content_type:
result["text"] = response.text_markdown # Convert HTML to markdown
else:
result["text"] = response.text
# Include cookies if present
if response.cookies:
result["cookies"] = response.cookies
return result
# Usage
result = fetch_and_process("https://httpbin.org/json")
print(f"Status: {result['status']}")
print(f"Data: {result['data']}")
Error Handling¶
The simplest way to turn a non-2xx response into an exception is
raise_for_status(). It raises HTTPStatusError for any non-2xx status and
otherwise returns the response, so it can be chained:
import httpr
try:
response = httpr.get("https://httpbin.org/status/500")
response.raise_for_status()
data = response.json()
except httpr.HTTPStatusError as exc:
print(f"HTTP Error: {exc}")
except httpr.HTTPError as exc:
print(f"Request failed: {exc}")
Note
Unlike requests, raise_for_status() treats any non-2xx response as
an error (including 1xx and 3xx), matching httpx. If you only want to act
on server/client errors, check response.is_error instead.
If you prefer to branch on the status yourself, use the status-class helpers:
import httpr
response = httpr.get("https://httpbin.org/status/500")
if response.is_error:
print(f"HTTP Error: {response.status_code} {response.reason_phrase}")
else:
data = response.json()
Streaming Responses¶
For large responses, you can stream the data instead of buffering it entirely in memory. This is useful for downloading large files, processing Server-Sent Events (SSE), or handling large API responses.
Basic Streaming¶
Use the stream() context manager to get a streaming response:
import httpr
client = httpr.Client()
# Stream response bytes
with client.stream("GET", "https://httpbin.org/stream-bytes/10000") as response:
print(f"Status: {response.status_code}")
for chunk in response.iter_bytes():
print(f"Received {len(chunk)} bytes")
# Process chunk without loading entire response in memory
Streaming Modes¶
httpr provides three ways to iterate over streaming responses:
1. Byte Chunks (iter_bytes())¶
Iterate over raw bytes chunks:
with client.stream("GET", "https://httpbin.org/stream-bytes/1000") as response:
for chunk in response.iter_bytes():
# chunk is bytes
process_binary_data(chunk)
Or use direct iteration (equivalent to iter_bytes()):
with client.stream("GET", "https://httpbin.org/stream-bytes/1000") as response:
for chunk in response: # Same as response.iter_bytes()
process_binary_data(chunk)
2. Text Chunks (iter_text())¶
Iterate over decoded text chunks:
with client.stream("GET", "https://httpbin.org/html") as response:
for text_chunk in response.iter_text():
# text_chunk is str, decoded using response encoding
print(text_chunk, end="")
The text is automatically decoded using the response's character encoding (from Content-Type header or detected from content).
3. Line by Line (iter_lines())¶
Iterate over the response line by line:
with client.stream("GET", "https://httpbin.org/stream/10") as response:
for line in response.iter_lines():
# line is str
print(line.strip())
This is particularly useful for:
- Server-Sent Events (SSE): Process events as they arrive
- JSONL/NDJSON: Parse newline-delimited JSON
- Log streaming: Process log lines in real-time
# Example: Processing Server-Sent Events
with client.stream("GET", "https://example.com/events") as response:
for line in response.iter_lines():
if line.startswith("data:"):
data = line[5:].strip() # Remove "data:" prefix
process_event(data)
Conditional Reading¶
You can check headers before deciding whether to read the body:
with client.stream("GET", "https://httpbin.org/get") as response:
# Headers are available immediately
content_type = response.headers.get("content-type")
content_length = response.headers.get("content-length")
if response.status_code != 200:
print(f"Error: {response.status_code}")
return # Don't read body
if content_length and int(content_length) > 1_000_000:
print("File too large!")
return # Don't read body
# Only read if checks pass
for chunk in response.iter_bytes():
process(chunk)
Reading All at Once¶
If you need to read the entire response after starting a stream:
with client.stream("GET", "https://httpbin.org/get") as response:
# Check headers first
if response.status_code == 200:
# Read entire remaining response
content = response.read()
print(f"Total size: {len(content)} bytes")
Downloading Large Files¶
Streaming is ideal for downloading large files:
import httpr
client = httpr.Client()
with client.stream("GET", "https://example.com/large-file.zip") as response:
if response.status_code == 200:
with open("large-file.zip", "wb") as f:
for chunk in response.iter_bytes():
f.write(chunk)
print("Download complete!")
With progress tracking:
with client.stream("GET", "https://example.com/large-file.zip") as response:
total_size = int(response.headers.get("content-length", 0))
downloaded = 0
with open("large-file.zip", "wb") as f:
for chunk in response.iter_bytes():
f.write(chunk)
downloaded += len(chunk)
if total_size:
percent = (downloaded / total_size) * 100
print(f"Downloaded: {percent:.1f}%", end="\r")
Streaming with POST¶
Streaming works with all HTTP methods:
with client.stream(
"POST",
"https://api.example.com/process",
json={"input": "data"}
) as response:
# Stream the API response
for line in response.iter_lines():
result = json.loads(line)
print(result)
Stream State¶
The streaming response tracks its state:
with client.stream("GET", "https://httpbin.org/get") as response:
print(response.is_closed) # False
print(response.is_consumed) # False
content = response.read()
print(response.is_consumed) # True (after reading)
print(response.is_closed) # True (after context manager exits)
Exception Handling¶
Streaming raises specific exceptions:
import httpr
try:
with client.stream("GET", "https://httpbin.org/get") as response:
content = response.read()
# This will raise StreamConsumed
more = response.read()
except httpr.StreamConsumed:
print("Cannot read stream twice")
except httpr.StreamClosed:
print("Stream was closed")
Async Streaming¶
AsyncClient.stream() yields an AsyncStreamingResponse, which is iterated with async for:
import asyncio
import httpr
async def stream_data():
async with httpr.AsyncClient() as client:
async with client.stream("GET", "https://httpbin.org/stream-bytes/1000") as response:
async for chunk in response.aiter_bytes():
process(chunk)
asyncio.run(stream_data())
aiter_bytes(), aiter_text(), aiter_lines() and aread() fetch each chunk on the client's thread pool, so the event loop keeps running other tasks (concurrent requests, timers, health checks) while the server is producing the next chunk. That matters for Server-Sent Events and LLM token streams, where a response can stay open for minutes:
async with client.stream("GET", "https://example.com/events") as response:
async for line in response.aiter_lines():
if line.startswith("data:"):
handle(line[5:].strip())
Note
The synchronous iter_bytes(), iter_text(), iter_lines() and read() are still available on the async response, but each step blocks the event loop until the next chunk arrives. Use the a-prefixed variants in async code.
Important Notes¶
- Always use context manager: The
withstatement ensures proper cleanup - Headers available immediately: You can check status, headers, and cookies before reading the body
- Cannot re-read: Once the stream is consumed, you cannot read it again
- Automatic cleanup: The stream is automatically closed when the context manager exits
Next Steps¶
- Authentication - Add authentication to requests
- Async Client - Use async/await for concurrent requests