Client API Reference

This page documents the main HttpClient class and related factory functions.

HttpClient

class HttpClient(config=None, session=None)

Production-ready HTTP client implementation following SOLID principles.

Parameters:

config (HttpClientConfig | None) – Client configuration object
session (requests.Session | None) – Optional requests Session to use

Thread Safety: This class is thread-safe. Multiple threads can safely share a single client instance.

Example:

from requestforge import HttpClient, HttpClientConfigBuilder

config = HttpClientConfigBuilder().with_base_url('https://api.example.com').build()
client = HttpClient(config)

response = client.get('/users')

Properties

HttpClient.adapter

HTTP adapter for connection pooling.

Type:: HTTPAdapter
Returns:: Configured HTTPAdapter instance

adapter = client.adapter
print(f"Pool connections: {adapter._pool_connections}")

HttpClient.session

Thread-safe requests session.

Type:: requests.Session
Returns:: Session instance for current thread

Note: Sessions are created lazily per thread and managed automatically.

session = client.session
print(session.headers)

HttpClient.auth_hook

Get the configured authentication hook.

Type:: AuthHookInterface | None
Returns:: Authentication hook if configured, None otherwise

if client.auth_hook:
    token = client.auth_hook.get_token()

HTTP Methods

HttpClient.get(url, params=None, headers=None, timeout=None, **kwargs)

Execute a GET request.

Parameters:

url (str) – URL path (relative or absolute)
params (dict) – URL query parameters
headers (dict) – Request headers
timeout (float) – Request timeout in seconds
kwargs – Additional request parameters

Returns:

HTTP response object

Return type:

HttpResponse

Raises:

HttpClientException – On request failure

Example:

# Simple GET
response = client.get('/users')

# With parameters
response = client.get('/users', params={'page': 1, 'limit': 10})

# With custom headers
response = client.get('/users', headers={'X-Custom': 'value'})

# With timeout
response = client.get('/slow-endpoint', timeout=60.0)

HttpClient.post(url, json_data=None, data=None, params=None, headers=None, timeout=None, **kwargs)

Execute a POST request.

Parameters:

url (str) – URL path
json_data (dict) – JSON data to send in request body
data (dict) – Form data to send in request body
params (dict) – URL query parameters
headers (dict) – Request headers
timeout (float) – Request timeout in seconds
kwargs – Additional request parameters

Returns:

HTTP response object

Return type:

HttpResponse

Raises:

HttpClientException – On request failure

Example:

# JSON POST
response = client.post('/users', json_data={
    'name': 'John Doe',
    'email': 'john@example.com'
})

# Form POST
response = client.post('/login', data={
    'username': 'john',
    'password': 'secret'
})

# With query parameters
response = client.post('/users', json_data={...}, params={'notify': 'true'})

HttpClient.put(url, json_data=None, data=None, params=None, headers=None, timeout=None, **kwargs)

Execute a PUT request.

Parameters:

url (str) – URL path
json_data (dict) – JSON data to send
data (dict) – Form data to send
params (dict) – URL query parameters
headers (dict) – Request headers
timeout (float) – Request timeout in seconds

Returns:

HTTP response object

Return type:

HttpResponse

Example:

response = client.put('/users/123', json_data={
    'name': 'Jane Doe',
    'email': 'jane@example.com'
})

HttpClient.patch(url, json_data=None, data=None, params=None, headers=None, timeout=None, **kwargs)

Execute a PATCH request.

Parameters:

url (str) – URL path
json_data (dict) – JSON data to send
data (dict) – Form data to send
params (dict) – URL query parameters
headers (dict) – Request headers
timeout (float) – Request timeout in seconds

Returns:

HTTP response object

Return type:

HttpResponse

Example:

# Partial update
response = client.patch('/users/123', json_data={
    'email': 'newemail@example.com'
})

HttpClient.delete(url, json_data=None, data=None, params=None, headers=None, timeout=None, **kwargs)

Execute a DELETE request.

Parameters:

url (str) – URL path
json_data (dict) – Optional JSON data
data (dict) – Optional form data
params (dict) – URL query parameters
headers (dict) – Request headers
timeout (float) – Request timeout in seconds

Returns:

HTTP response object

Return type:

HttpResponse

Example:

# Simple delete
response = client.delete('/users/123')

# Delete with reason
response = client.delete('/users/123', json_data={
    'reason': 'User requested account deletion'
})

Core Request Method

HttpClient.request(request)

Execute an HTTP request with comprehensive retry logic and lifecycle hooks.

Parameters:

request (HttpRequest) – HTTP request object

Returns:

HTTP response object

Return type:

HttpResponse

Raises:

MaxRetryException – If max retries exceeded
TimeoutException – On request timeout
ConnectionException – On connection failure
HttpStatusException – On HTTP error status
AuthenticationException – On authentication failure

This is the core method that all other HTTP methods delegate to. It handles:

Request hook execution
Authentication injection
Request execution
Response hook execution
Error handling and retry logic
Authentication retry on 401

Example:

from requestforge import HttpRequest, HttpMethod

request = HttpRequest(
    method=HttpMethod.POST,
    url='/users',
    json_data={'name': 'John'},
    headers={'X-Custom': 'value'},
    timeout=30.0
)

response = client.request(request)

Request Flow:

1. Execute request hooks (before_request)
2. Execute auth hook (inject token)
3. Execute HTTP request
4. Check for auth error (401)
   ├─ Yes → Refresh token & retry
   └─ No → Continue
5. Check for retryable error
   ├─ Yes → Retry with backoff
   └─ No → Continue
6. Execute response hooks (after_response)
7. Return response

Concurrent Requests

HttpClient.request_many(requests_list, max_workers=5, fail_fast=False)

Execute multiple HTTP requests concurrently.

Parameters:

requests_list (list) – List of HttpRequest objects
max_workers (int) – Maximum number of concurrent worker threads
fail_fast (bool) – If True, stop on first error

Returns:

Generator yielding (index, result) tuples

Return type:

Generator[tuple[int, HttpResponse | HttpClientException], None, None]

Raises:

HttpClientException – If fail_fast=True and request fails

Results are yielded as they complete (not in order). Each result is a tuple of:

index (int): Index of request in original list
result (HttpResponse | HttpClientException): Response or exception

Example:

from requestforge import HttpRequest, HttpMethod

requests = [
    HttpRequest(method=HttpMethod.GET, url=f'/users/{i}')
    for i in range(1, 11)
]

# Execute with 5 workers
for index, result in client.request_many(requests, max_workers=5):
    if isinstance(result, HttpResponse):
        print(f"Request {index}: Success")
    else:
        print(f"Request {index}: Failed - {result}")

Fail-Fast Example:

try:
    results = client.request_many(requests, max_workers=10, fail_fast=True)
    for index, response in results:
        process(response)
except HttpClientException as e:
    print(f"Batch failed: {e}")

Resource Management

HttpClient.close()

Close the client and release all resources.

This method:

Closes thread-local sessions
Closes shared session (if provided)
Releases connection pool resources
Marks client as closed

client = HttpClient(config)
try:
    response = client.get('/users')
finally:
    client.close()

HttpClient.__enter__()

Enter context manager.

Returns:: Self
Return type:: HttpClient

HttpClient.__exit__(exc_type, exc_val, exc_tb)

Exit context manager and cleanup resources.

Example:

with HttpClient(config) as client:
    response = client.get('/users')
# Client automatically closed

Factory Functions

create_client

create_client(base_url='', timeout=30.0, max_retries=3, headers=None, enable_logging=True, auth_hook=None)

Factory function to create a configured HTTP client.

Parameters:

base_url (str) – Base URL for all requests
timeout (float) – Default timeout in seconds
max_retries (int) – Maximum retry attempts
headers (dict) – Default headers
enable_logging (bool) – Enable logging hooks
auth_hook (AuthHookInterface) – Authentication hook

Returns:

Configured HTTP client

Return type:

HttpClient

Example:

from requestforge import create_client

# Simple client
client = create_client('https://api.example.com')

# With custom configuration
client = create_client(
    base_url='https://api.example.com',
    timeout=60.0,
    max_retries=5,
    headers={'User-Agent': 'MyApp/1.0'},
    enable_logging=True
)

# With authentication
from requestforge import TokenManager

client = create_client(
    base_url='https://api.example.com',
    auth_hook=TokenAuthHook(token_manager)
)

requestforge (Context Manager)

http_client(base_url='', **kwargs)

Context manager factory for HTTP client.

Parameters:

base_url (str) – Base URL for requests
kwargs – Additional configuration (same as create_client)

Returns:

Context manager yielding HttpClient

Return type:

ContextManager[HttpClient]

Example:

from requestforge import http_client

with http_client('https://api.example.com') as client:
    response = client.get('/users')
    users = response.json()
# Client automatically closed

With Configuration:

with http_client(
    base_url='https://api.example.com',
    timeout=60.0,
    max_retries=5
) as client:
    response = client.get('/data')

Complete Example

from requestforge import (
    HttpClient,
    HttpClientConfigBuilder,
    HttpRequest,
    HttpMethod,
    create_client,
    requestforge
)

# Method 1: Using builder pattern
config = (
    HttpClientConfigBuilder()
    .with_base_url('https://api.example.com')
    .with_timeout(30.0)
    .with_retry(max_retries=3)
    .with_bearer_token('your-token')
    .with_logging()
    .build()
)

client = HttpClient(config)

# Simple requests
response = client.get('/users')
response = client.post('/users', json_data={'name': 'John'})
response = client.put('/users/1', json_data={'name': 'Jane'})
response = client.patch('/users/1', json_data={'email': 'new@email.com'})
response = client.delete('/users/1')

# Advanced request
request = HttpRequest(
    method=HttpMethod.POST,
    url='/users',
    headers={'X-Custom': 'value'},
    json_data={'name': 'John'},
    timeout=60.0
)
response = client.request(request)

# Concurrent requests
requests = [
    HttpRequest(method=HttpMethod.GET, url=f'/users/{i}')
    for i in range(1, 11)
]
for index, result in client.request_many(requests, max_workers=5):
    print(f"Request {index}: {result.status_code}")

client.close()

# Method 2: Using factory function
client = create_client(
    base_url='https://api.example.com',
    timeout=30.0,
    max_retries=3
)

response = client.get('/users')
client.close()

# Method 3: Using context manager
with http_client('https://api.example.com') as client:
    response = client.get('/users')

Client API Reference

HttpClient

Properties

HTTP Methods

Core Request Method

Concurrent Requests

Resource Management

Factory Functions

create_client

requestforge (Context Manager)

Complete Example

See Also