Skip to content
Closed
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 2 additions & 0 deletions httpx/__init__.py
Original file line number Diff line number Diff line change
Expand Up @@ -10,6 +10,7 @@
from ._transports import *
from ._types import *
from ._urls import *
from ._utils import normalize_header_key

try:
from ._main import main
Expand Down Expand Up @@ -63,6 +64,7 @@ def main() -> None: # type: ignore
"MockTransport",
"NetRCAuth",
"NetworkError",
"normalize_header_key",
"options",
"patch",
"PoolTimeout",
Expand Down
121 changes: 121 additions & 0 deletions httpx/_models.py
Original file line number Diff line number Diff line change
Expand Up @@ -831,6 +831,127 @@ def raise_for_status(self) -> Response:
def json(self, **kwargs: typing.Any) -> typing.Any:
return jsonlib.loads(self.content, **kwargs)

def json_or_text(self, **kwargs: typing.Any) -> typing.Any:
"""
Attempt to parse response as JSON, falling back to text if parsing fails.

This is a convenience method that combines the behavior of `json()` and `text`.
It first attempts to parse the response body as JSON. If that fails due to
invalid JSON or an empty response, it returns the text content instead.

Args:
**kwargs: Additional arguments passed to json.loads() when parsing JSON.

Returns:
The parsed JSON data if successful, otherwise the text content of the response.

Example:
>>> response = httpx.get("https://api.example.com/data")
>>> data = response.json_or_text() # Returns JSON dict or text string
>>> # Useful for APIs that may return either JSON or plain text
>>> response = httpx.get("https://example.com/endpoint")
>>> content = response.json_or_text() # Handles both content types gracefully

Note:
This method is particularly useful when:
- Working with APIs that may return different content types
- You want automatic fallback behavior without exception handling
- Processing responses where the content type is uncertain
"""
try:
return jsonlib.loads(self.content, **kwargs)
except (jsonlib.JSONDecodeError, UnicodeDecodeError, ValueError):
return self.text

def save_to_file(
self,
path: str,
mode: str = "auto",
*,
encoding: str | None = None,
indent: int | None = None,
) -> None:
"""
Save the response content to a file.

This method provides a convenient way to save response content to disk with
automatic handling of different content types based on the mode parameter.

Args:
path: The file path where the content should be saved.
mode: The save mode, one of:
- 'auto': Automatically detect based on Content-Type header
(saves as JSON if content-type is application/json, binary otherwise)
- 'binary': Save raw bytes (use for images, PDFs, etc.)
- 'text': Save as text using the response's encoding
- 'json': Parse as JSON and save with formatting
encoding: Text encoding to use when mode is 'text'. Defaults to the
response's detected encoding. Only applicable for 'text' mode.
indent: Number of spaces for JSON indentation when mode is 'json'.
Defaults to 2. Only applicable for 'json' mode.

Raises:
ValueError: If an invalid mode is specified.
OSError: If there are issues writing to the file.
JSONDecodeError: If mode is 'json' but content is not valid JSON.

Example:
>>> # Save binary content (image, PDF, etc.)
>>> response = httpx.get("https://example.com/image.png")
>>> response.save_to_file("image.png", mode="binary")
>>>
>>> # Save JSON with formatting
>>> response = httpx.get("https://api.example.com/data")
>>> response.save_to_file("data.json", mode="json", indent=4)
>>>
>>> # Auto-detect based on content type
>>> response = httpx.get("https://example.com/file")
>>> response.save_to_file("output.txt", mode="auto")
>>>
>>> # Save as text with specific encoding
>>> response = httpx.get("https://example.com/page")
>>> response.save_to_file("page.html", mode="text", encoding="utf-8")

Note:
- The 'auto' mode checks the Content-Type header to determine format
- For 'binary' and 'json' modes, the file is written in binary mode
- For 'text' mode, the file is written in text mode with specified encoding
- Parent directories are not created automatically; they must exist
"""
import pathlib

file_path = pathlib.Path(path)

if mode not in ("auto", "binary", "text", "json"):
raise ValueError(
f"Invalid mode '{mode}'. Must be one of: 'auto', 'binary', 'text', 'json'"
)

# Determine actual mode if auto
if mode == "auto":
content_type = self.headers.get("content-type", "").lower()
if "application/json" in content_type:
mode = "json"
elif any(
t in content_type
for t in ["text/", "application/xml", "application/javascript"]
):
mode = "text"
else:
mode = "binary"

# Save based on determined mode
if mode == "binary":
file_path.write_bytes(self.content)
elif mode == "text":
text_encoding = encoding or self.encoding or "utf-8"
file_path.write_text(self.text, encoding=text_encoding)
elif mode == "json":
json_data = self.json()
json_indent = 2 if indent is None else indent
formatted_json = jsonlib.dumps(json_data, indent=json_indent, ensure_ascii=False)
file_path.write_text(formatted_json, encoding="utf-8")

@property
def cookies(self) -> Cookies:
if not hasattr(self, "_cookies"):
Expand Down
49 changes: 49 additions & 0 deletions httpx/_utils.py
Original file line number Diff line number Diff line change
Expand Up @@ -227,6 +227,10 @@ def __eq__(self, other: typing.Any) -> bool:


def is_ipv4_hostname(hostname: str) -> bool:
"""
Check if the given hostname is a valid IPv4 address.
Supports CIDR notation by checking only the address part.
"""
try:
ipaddress.IPv4Address(hostname.split("/")[0])
except Exception:
Expand All @@ -235,8 +239,53 @@ def is_ipv4_hostname(hostname: str) -> bool:


def is_ipv6_hostname(hostname: str) -> bool:
"""
Check if the given hostname is a valid IPv6 address.
Supports CIDR notation by checking only the address part.
"""
try:
ipaddress.IPv6Address(hostname.split("/")[0])
except Exception:
return False
return True


def is_ip_address(hostname: str) -> bool:
"""
Check if the given hostname is a valid IP address (either IPv4 or IPv6).
Supports CIDR notation by checking only the address part.
"""
return is_ipv4_hostname(hostname) or is_ipv6_hostname(hostname)


def normalize_header_key(key: str, *, preserve_case: bool = False) -> str:
"""
Normalize HTTP header keys for consistent comparison and storage.

By default, converts header keys to lowercase following HTTP/2 conventions.
Can optionally preserve the original case for HTTP/1.1 compatibility.

Args:
key: The header key to normalize
preserve_case: If True, preserve the original case. If False (default),
convert to lowercase.

Returns:
The normalized header key as a string

Examples:
>>> normalize_header_key("Content-Type")
'content-type'
>>> normalize_header_key("Content-Type", preserve_case=True)
'Content-Type'
>>> normalize_header_key("X-Custom-Header")
'x-custom-header'

Note:
This function is useful when working with HTTP headers across different
protocol versions. HTTP/2 requires lowercase header names, while HTTP/1.1
traditionally uses title-case headers (though comparison is case-insensitive).
"""
if preserve_case:
return key.strip()
return key.strip().lower()
Loading