Models API Reference

This page documents the data transfer objects (DTOs) used for HTTP operations.

All models are immutable (frozen dataclasses) following the Value Object pattern.

HttpMethod

class HttpMethod

Enumeration of HTTP methods.

Values:

GET: str = 'GET'

POST: str = 'POST'

PUT: str = 'PUT'

PATCH: str = 'PATCH'

DELETE: str = 'DELETE'

HEAD: str = 'HEAD'

OPTIONS: str = 'OPTIONS'

Example:

from requestforge import HttpMethod

method = HttpMethod.GET
print(method.value)  # 'GET'

# Can be used in comparisons
if method == HttpMethod.GET:
    print("It's a GET request")

HttpRequest

class HttpRequest(method, url, headers=None, params=None, data=None, json_data=None, files=None, timeout=None, auth=None)

Immutable HTTP request representation.

Parameters:

method (HttpMethod) – HTTP method
url (str) – Request URL (relative or absolute)
headers (dict) – Request headers (optional)
params (dict) – URL query parameters (optional)
data (dict) – Form data (optional)
json_data (dict) – JSON data (optional)
files (dict) – Files to upload (optional)
timeout (float) – Request timeout in seconds (optional)
auth (tuple) – Authentication tuple (username, password) (optional)

Note: This is a frozen dataclass. Once created, it cannot be modified. Use helper methods to create modified copies.

Attributes

HttpRequest.method: HttpMethod: HTTP method for the request.

HttpRequest.url: str: Request URL (can be relative or absolute).

HttpRequest.headers: dict[str, str] | None: Request headers.

HttpRequest.params: dict[str, Any] | None: URL query parameters.

HttpRequest.data: dict[str, Any] | None: Form-encoded data for request body.

HttpRequest.json_data: dict[str, Any] | None: JSON data for request body.

HttpRequest.files: dict[str, Any] | None: Files to upload (multipart/form-data).

HttpRequest.timeout: float | None: Request timeout in seconds.

HttpRequest.auth: tuple | None: Authentication credentials (username, password).

Methods

HttpRequest.with_headers(headers)

Create new request with additional headers.

Parameters:: headers (dict) – Headers to add/override
Returns:: New HttpRequest instance with merged headers
Return type:: HttpRequest

Example:

request = HttpRequest(method=HttpMethod.GET, url='/users')

# Add headers (returns new instance)
authenticated = request.with_headers({
    'Authorization': 'Bearer token123',
    'X-Custom': 'value'
})

print(request.headers)         # None (original unchanged)
print(authenticated.headers)   # {'Authorization': '...', 'X-Custom': '...'}

HttpRequest.to_dict()

Convert request to dictionary.

Returns:: Dictionary representation of request
Return type:: dict[str, Any]

Example:

request = HttpRequest(
    method=HttpMethod.POST,
    url='/users',
    json_data={'name': 'John'}
)

data = request.to_dict()
# {
#     'method': HttpMethod.POST,
#     'url': '/users',
#     'headers': None,
#     'params': None,
#     'data': None,
#     'json_data': {'name': 'John'},
#     'files': None,
#     'timeout': None,
#     'auth': None
# }

HttpRequest.to_curl()

Convert request to cURL command string.

Returns:: cURL command that represents this request
Return type:: str

Example:

request = HttpRequest(
    method=HttpMethod.POST,
    url='https://api.example.com/users',
    headers={'Authorization': 'Bearer token'},
    json_data={'name': 'John'}
)

print(request.to_curl())
# curl -X 'POST' -H 'Authorization: Bearer token' \
#      -H 'Content-Type: application/json' \
#      --data '{"name": "John"}' \
#      'https://api.example.com/users'

Examples

from requestforge import HttpRequest, HttpMethod

# Simple GET
request = HttpRequest(method=HttpMethod.GET, url='/users')

# GET with parameters
request = HttpRequest(
    method=HttpMethod.GET,
    url='/users',
    params={'page': 1, 'limit': 10}
)

# POST with JSON
request = HttpRequest(
    method=HttpMethod.POST,
    url='/users',
    json_data={'name': 'John', 'email': 'john@example.com'},
    headers={'Content-Type': 'application/json'}
)

# POST with form data
request = HttpRequest(
    method=HttpMethod.POST,
    url='/login',
    data={'username': 'john', 'password': 'secret'}
)

# File upload
request = HttpRequest(
    method=HttpMethod.POST,
    url='/upload',
    files={'document': open('file.pdf', 'rb')}
)

# With timeout
request = HttpRequest(
    method=HttpMethod.GET,
    url='/slow-endpoint',
    timeout=60.0
)

# With basic auth
request = HttpRequest(
    method=HttpMethod.GET,
    url='/protected',
    auth=('username', 'password')
)

HttpResponse

class HttpResponse(status_code, headers, content, elapsed_ms, url, request, encoding=None)

Immutable HTTP response representation.

Parameters:

status_code (int) – HTTP status code
headers (dict) – Response headers
content (bytes) – Response body content
elapsed_ms (float) – Request duration in milliseconds
url (str) – Final URL (after redirects)
request (HttpRequest) – Original request
encoding (str) – Response encoding (optional)

Attributes

HttpResponse.status_code: int: HTTP status code (e.g., 200, 404, 500).

HttpResponse.headers: dict[str, str]: Response headers.

HttpResponse.content: bytes: Raw response body content.

HttpResponse.elapsed_ms: float: Request duration in milliseconds.

HttpResponse.url: str: Final URL after any redirects.

HttpResponse.request: HttpRequest: Original request that generated this response.

HttpResponse.encoding: str | None: Response encoding (e.g., ‘utf-8’).

Properties

HttpResponse.text: str

Decoded response content as string.

Example:

response = client.get('/users')
print(response.text)  # Decoded string

HttpResponse.is_success: bool

True if status code is 2xx (200-299).

Example:

response = client.get('/users')
if response.is_success:
    print("Request succeeded")

HttpResponse.is_client_error: bool: True if status code is 4xx (400-499).

HttpResponse.is_server_error: bool: True if status code is 5xx (500-599).

Methods

HttpResponse.json()

Parse response content as JSON.

Returns:: Parsed JSON data
Return type:: Any
Raises:: ResponseParseException – If content is not valid JSON

Example:

response = client.get('/users/1')
user = response.json()
print(user['name'])

HttpResponse.json_or_none()

Parse response as JSON, return None on failure.

Returns:: Parsed JSON data or None
Return type:: Any | None

Example:

response = client.get('/data')
data = response.json_or_none()

if data is not None:
    process(data)
else:
    print("Response is not JSON")

Examples

from requestforge import HttpClient

client = create_client('https://api.example.com')
response = client.get('/users/1')

# Status code
print(response.status_code)  # 200

# Check status
if response.is_success:
    print("Success!")
elif response.is_client_error:
    print("Client error (4xx)")
elif response.is_server_error:
    print("Server error (5xx)")

# Headers
print(response.headers['Content-Type'])
print(response.headers.get('X-Rate-Limit'))

# Body content
print(response.content)      # bytes
print(response.text)         # str

# Parse JSON
user = response.json()
print(user['name'])

# Safe JSON parsing
data = response.json_or_none()
if data:
    print(data)

# Metadata
print(f"Request took {response.elapsed_ms}ms")
print(f"Final URL: {response.url}")
print(f"Encoding: {response.encoding}")

# Access original request
print(response.request.url)
print(response.request.method)

RequestContext

class RequestContext(request, attempt=0, max_retries=0, metadata=None)

Mutable context for request lifecycle.

Parameters:

request (HttpRequest) – The HTTP request
attempt (int) – Current attempt number (0-indexed)
max_retries (int) – Maximum retry attempts
metadata (dict) – Additional context metadata

Note: Unlike HttpRequest/HttpResponse, RequestContext is mutable to track state during retries.

Attributes

RequestContext.request: HttpRequest: The HTTP request being executed.

RequestContext.attempt: int: Current attempt number (0 = first attempt, 1 = first retry, etc.).

RequestContext.max_retries: int: Maximum number of retry attempts allowed.

RequestContext.metadata: dict[str, Any]: Additional context data (e.g., correlation IDs, timing data).

Properties

RequestContext.is_retry: bool

True if this is a retry attempt (attempt > 0).

Example:

if context.is_retry:
    print(f"Retry attempt {context.attempt}")

Methods

RequestContext.increment_attempt()

Increment the attempt counter.

Example:

context.increment_attempt()
print(context.attempt)  # Incremented

Examples

from requestforge.models import RequestContext, HttpRequest, HttpMethod

request = HttpRequest(method=HttpMethod.GET, url='/users')
context = RequestContext(
    request=request,
    attempt=0,
    max_retries=3,
    metadata={}
)

# Check if retry
if context.is_retry:
    print(f"Retry #{context.attempt}")
else:
    print("First attempt")

# Store metadata
context.metadata['request_id'] = 'abc123'
context.metadata['start_time'] = time.time()

# Increment attempt
context.increment_attempt()
print(f"Attempt: {context.attempt}")  # 1

# Check retry status
print(context.is_retry)  # True

FetchContext

class FetchContext(tokens=None, metadata=None)

Context passed through the token fetch pipeline.

Parameters:

tokens (dict) – Tokens fetched by previous steps
metadata (dict) – Additional metadata

Used in multi-step authentication pipelines to pass tokens between fetcher steps.

Attributes

FetchContext.tokens: dict[str, TokenData]: Dictionary mapping step names to their fetched tokens.

FetchContext.metadata: dict[str, Any]: Additional context metadata.

Methods

FetchContext.get_token(name)

Get token by step name.

Parameters:: name (str) – Step name
Returns:: Token data or None
Return type:: TokenData | None

FetchContext.get_token_value(name)

Get token access_token value by step name.

Parameters:: name (str) – Step name
Returns:: Access token string or None
Return type:: str | None

FetchContext.add_token(name, token)

Add token to context.

Parameters:

name (str) – Step name
token (TokenData) – Token data

FetchContext.has_token(name)

Check if token exists in context.

Parameters:: name (str) – Step name
Returns:: True if token exists
Return type:: bool

FetchContext.has_valid_token(name)

Check if token exists and is not expired.

Parameters:: name (str) – Step name
Returns:: True if valid token exists
Return type:: bool

Examples

from requestforge.models import FetchContext
from requestforge.config import TokenData

context = FetchContext()

# Add token
app_token = TokenData(access_token='app-token-123', token_type='Bearer')
context.add_token('app_token', app_token)

# Check if token exists
if context.has_token('app_token'):
    print("App token available")

# Get token
token = context.get_token('app_token')
print(token.access_token)

# Get just the token value
token_value = context.get_token_value('app_token')
print(token_value)  # 'app-token-123'

# Check validity
if context.has_valid_token('app_token'):
    print("Token is valid and not expired")

# Store metadata
context.metadata['pipeline_start'] = time.time()

Complete Example

from requestforge import HttpClient, HttpRequest, HttpMethod

# Create request
request = HttpRequest(
    method=HttpMethod.POST,
    url='https://api.example.com/users',
    headers={'Authorization': 'Bearer token'},
    json_data={'name': 'John', 'email': 'john@example.com'},
    timeout=30.0
)

# Execute request
client = HttpClient(config)
response = client.request(request)

# Check response
print(f"Status: {response.status_code}")
print(f"Success: {response.is_success}")
print(f"Duration: {response.elapsed_ms}ms")

# Parse response
if response.is_success:
    user = response.json()
    print(f"Created user: {user['id']}")
else:
    print(f"Error: {response.text}")

# Convert to cURL for debugging
print(request.to_curl())

Models API Reference

HttpMethod

HttpRequest

Attributes

Methods

Examples

HttpResponse

Attributes

Properties

Methods

Examples

RequestContext

Attributes

Properties

Methods

Examples

FetchContext

Attributes

Methods

Examples

Complete Example

See Also