Design Principles
Request Forge is built following industry-standard design principles and patterns to ensure maintainability, extensibility, and robustness.
SOLID Principles
Single Responsibility Principle (SRP)
Definition: A class should have only one reason to change.
Implementation:
Each class in Request Forge has a single, well-defined responsibility:
HttpClient- Execute HTTP requestsHttpClientConfig- Hold configuration dataTokenManager- Manage token lifecycleRetryStrategy- Determine retry behaviorTokenStorage- Store and retrieve tokens
Example:
# Good ✅ - Single responsibility
class TokenManager:
"""Manages token lifecycle: fetch, cache, refresh."""
def get_token(self): ...
def invalidate_token(self): ...
def force_refresh(self): ...
class TokenStorage:
"""Stores tokens: get, set, delete."""
def get(self, key): ...
def set(self, key, token): ...
def delete(self, key): ...
# Avoid ❌ - Multiple responsibilities
class TokenManagerAndStorage:
"""Manages tokens AND stores them."""
def get_token(self): ...
def invalidate_token(self): ...
def _store_token(self): ...
def _retrieve_token(self): ...
Benefits:
Easier to test (mock single responsibility)
Easier to maintain (changes isolated)
Better code organization
Open/Closed Principle (OCP)
Definition: Software entities should be open for extension but closed for modification.
Implementation:
Request Forge uses interfaces and abstract base classes to allow extension without modification:
# Base interface (closed for modification)
class RetryStrategyInterface(ABC):
@abstractmethod
def should_retry(self, context, exception) -> bool: ...
@abstractmethod
def get_delay(self, context) -> float: ...
# Built-in implementations
class ExponentialBackoffRetryStrategy(RetryStrategyInterface): ...
class CircuitBreakerRetryStrategy(RetryStrategyInterface): ...
# Custom implementation (extension without modification)
class MyCustomRetryStrategy(RetryStrategyInterface):
def should_retry(self, context, exception) -> bool:
# Custom logic
return my_custom_logic(exception)
def get_delay(self, context) -> float:
return calculate_my_delay(context)
Extension Points:
Retry strategies via
RetryStrategyInterfaceHooks via
RequestHookInterface,ResponseHookInterface,ErrorHookInterfaceToken providers via
TokenProviderInterfaceToken storage via
TokenStorageInterfaceToken fetchers via
TokenFetcherInterface
Benefits:
Add new behavior without changing existing code
Maintain backward compatibility
Reduce risk of breaking existing functionality
Liskov Substitution Principle (LSP)
Definition: Objects of a superclass should be replaceable with objects of a subclass without breaking the application.
Implementation:
All implementations of an interface can be used interchangeably:
# Any TokenStorageInterface can be substituted
def create_token_manager(storage: TokenStorageInterface):
return TokenManager(provider, storage)
# All these work identically from caller's perspective
manager1 = create_token_manager(InMemoryTokenStorage())
manager2 = create_token_manager(DjangoCacheTokenStorage())
manager3 = create_token_manager(RedisTokenStorage())
# All behave the same
token1 = manager1.get_token()
token2 = manager2.get_token()
token3 = manager3.get_token()
Contract Preservation:
class RetryStrategyInterface:
"""Contract: should_retry returns bool, get_delay returns non-negative float."""
@abstractmethod
def should_retry(self, context, exception) -> bool: ...
@abstractmethod
def get_delay(self, context) -> float: ...
# Good ✅ - Preserves contract
class CustomRetryStrategy(RetryStrategyInterface):
def should_retry(self, context, exception) -> bool:
return True # Returns bool
def get_delay(self, context) -> float:
return 1.0 # Returns float >= 0
# Bad ❌ - Violates contract
class BadRetryStrategy(RetryStrategyInterface):
def should_retry(self, context, exception) -> bool:
return "yes" # Returns string, not bool!
def get_delay(self, context) -> float:
return -1.0 # Returns negative, violates contract!
Benefits:
Polymorphism works correctly
Easier to swap implementations
Predictable behavior
Interface Segregation Principle (ISP)
Definition: No client should be forced to depend on methods it does not use.
Implementation:
Request Forge uses focused, specific interfaces rather than large, monolithic ones:
# Good ✅ - Focused interfaces
class RequestHookInterface:
"""Only for request modification."""
def before_request(self, request, context): ...
class ResponseHookInterface:
"""Only for response processing."""
def after_response(self, response, context): ...
class ErrorHookInterface:
"""Only for error handling."""
def on_error(self, exception, context): ...
# Avoid ❌ - Fat interface
class HookInterface:
"""Forces implementation of all methods even if not needed."""
def before_request(self, request, context): ...
def after_response(self, response, context): ...
def on_error(self, exception, context): ...
Specialized Interfaces:
# Implement only what you need
class LoggingRequestHook(RequestHookInterface):
"""Only implements before_request."""
def before_request(self, request, context):
logger.info(f"Request: {request.url}")
class MetricsResponseHook(ResponseHookInterface):
"""Only implements after_response."""
def after_response(self, response, context):
metrics.timing('request_duration', response.elapsed_ms)
Benefits:
Simpler implementations
Clearer intent
Less coupling
Dependency Inversion Principle (DIP)
Definition: High-level modules should not depend on low-level modules. Both should depend on abstractions.
Implementation:
Request Forge depends on interfaces, not concrete implementations:
# High-level class depends on abstractions
class HttpClient:
def __init__(
self,
config: HttpClientConfig, # Concrete config is fine (data)
session: requests.Session | None = None
):
self._config = config
self._session = session
# Depends on interface, not concrete implementation
self._retry_strategy: RetryStrategyInterface = config.retry_strategy
self._auth_hook: AuthHookInterface = config.auth_hook
# Good ✅ - Injection of abstraction
class TokenManager:
def __init__(
self,
provider: TokenProviderInterface, # Abstraction
storage: TokenStorageInterface # Abstraction
):
self._provider = provider
self._storage = storage
# Bad ❌ - Direct dependency on concrete class
class TokenManager:
def __init__(self):
self._provider = ClientCredentialsTokenProvider(...) # Concrete
self._storage = InMemoryTokenStorage() # Concrete
Dependency Injection:
# Dependencies injected, not created internally
provider = ClientCredentialsTokenProvider(...)
storage = DjangoCacheTokenStorage()
token_manager = TokenManager(provider, storage)
config = (
HttpClientConfigBuilder()
.with_token_auth(token_manager=token_manager)
.with_retry_strategy(ExponentialBackoffRetryStrategy())
.build()
)
Benefits:
Easy to test (inject mocks)
Easy to swap implementations
Loose coupling
Design Patterns
Builder Pattern
Purpose: Separate construction of complex objects from their representation.
Implementation: HttpClientConfigBuilder
# Fluent builder for complex configuration
config = (
HttpClientConfigBuilder()
.with_base_url('https://api.example.com')
.with_timeout(30.0)
.with_retry(max_retries=3)
.with_bearer_token('token')
.with_logging()
.build() # Returns immutable HttpClientConfig
)
Benefits:
Readable, fluent API
Immutable result
Validation at build time
Optional parameters easy to handle
Strategy Pattern
Purpose: Define a family of algorithms, encapsulate each one, and make them interchangeable.
Implementation: Retry strategies
# Strategy interface
class RetryStrategyInterface:
def should_retry(self, context, exception) -> bool: ...
def get_delay(self, context) -> float: ...
# Concrete strategies
class NoRetryStrategy(RetryStrategyInterface): ...
class SimpleRetryStrategy(RetryStrategyInterface): ...
class ExponentialBackoffRetryStrategy(RetryStrategyInterface): ...
class CircuitBreakerRetryStrategy(RetryStrategyInterface): ...
# Context uses strategy
class HttpClient:
def __init__(self, config):
self._retry_strategy = config.retry_strategy
def request(self, request):
# Strategy determines retry behavior
if self._retry_strategy.should_retry(context, exception):
delay = self._retry_strategy.get_delay(context)
# Retry logic
Benefits:
Runtime algorithm selection
Easy to add new strategies
Testable in isolation
Chain of Responsibility Pattern
Purpose: Pass request along a chain of handlers.
Implementation: Hooks pipeline
# Chain of request hooks
request_hooks = [
CorrelationIdHook(),
LoggingRequestHook(),
CustomHeaderHook()
]
# Execute chain
for hook in request_hooks:
request = hook.before_request(request, context)
# Chain of response hooks
response_hooks = [
RateLimitResponseHook(),
LoggingResponseHook(),
MetricsHook()
]
for hook in response_hooks:
response = hook.after_response(response, context)
Benefits:
Flexible request/response processing
Easy to add/remove handlers
Each handler independent
Factory Pattern
Purpose: Create objects without specifying exact class.
Implementation: Client factory functions
def create_client(
base_url: str = '',
timeout: float = 30.0,
max_retries: int = 3,
**kwargs
) -> HttpClient:
"""Factory function creates configured client."""
builder = (
HttpClientConfigBuilder()
.with_base_url(base_url)
.with_timeout(timeout)
.with_retry(max_retries=max_retries)
)
# Additional configuration
if kwargs.get('headers'):
builder.with_headers(kwargs['headers'])
return HttpClient(builder.build())
Benefits:
Simple client creation
Encapsulates configuration complexity
Consistent defaults
Template Method Pattern
Purpose: Define skeleton of algorithm, let subclasses override specific steps.
Implementation: Token fetchers
class HttpTokenFetcher(TokenFetcherInterface):
"""Template for HTTP token fetching."""
def fetch(self, context=None) -> TokenData:
# Template method
headers = self._build_request_headers(context) # Hook
data = self._build_request_data(context) # Hook
response = self._make_request(headers, data) # Common
return self._parse_response(response) # Hook
def _build_request_headers(self, context):
"""Override in subclass."""
return {}
def _build_request_data(self, context):
"""Override in subclass."""
return {}
def _parse_response(self, response):
"""Override in subclass."""
raise NotImplementedError
# Subclass overrides specific steps
class CustomTokenFetcher(HttpTokenFetcher):
def _build_request_headers(self, context):
headers = super()._build_request_headers(context)
if context and 'app_token' in context:
headers['X-App-Token'] = context['app_token'].access_token
return headers
Benefits:
Code reuse
Consistent structure
Easy customization
Value Object Pattern
Purpose: Immutable objects representing descriptive aspects.
Implementation: All models (HttpRequest, HttpResponse, TokenData, etc.)
@dataclass(frozen=True)
class HttpRequest:
"""Immutable request representation."""
method: HttpMethod
url: str
headers: dict | None = None
# ... other fields
def with_headers(self, headers: dict) -> HttpRequest:
"""Returns new instance with merged headers."""
merged = {**(self.headers or {}), **headers}
return HttpRequest(
method=self.method,
url=self.url,
headers=merged,
# ... copy other fields
)
Benefits:
Thread-safe (immutable)
No side effects
Easy to reason about
Immutability
Principle: Data objects should not change after creation.
Implementation:
All data transfer objects (DTOs) are immutable:
# All frozen dataclasses
@dataclass(frozen=True)
class HttpRequest: ...
@dataclass(frozen=True)
class HttpResponse: ...
@dataclass(frozen=True)
class HttpClientConfig: ...
@dataclass
class TokenData: ... # Effectively immutable in practice
Benefits:
Thread-safe
Prevents accidental mutations
Makes data flow explicit
Easier to debug
Modification Pattern:
# Cannot modify directly
request = HttpRequest(method=HttpMethod.GET, url='/users')
# request.headers = {'X-Custom': 'value'} # ❌ Error: frozen
# Create new instance with changes
modified = request.with_headers({'X-Custom': 'value'}) # ✅
Separation of Concerns
Principle: Different concerns should be in different modules/classes.
Implementation:
requestforge/
├── client.py # HTTP execution
├── config.py # Configuration
├── models.py # Data models
├── retry.py # Retry logic
├── hooks.py # Lifecycle hooks
├── token_manager.py # Token management
├── pipelines.py # Multi-step auth
├── fetcher.py # Token fetching
├── exceptions.py # Error types
└── utils.py # Utilities
Concerns:
HTTP execution: client.py
Configuration: config.py
Data representation: models.py
Retry logic: retry.py
Request/response lifecycle: hooks.py
Token management: token_manager.py, pipelines.py, fetcher.py
Error handling: exceptions.py
Utilities: utils.py
Benefits:
Easy to navigate codebase
Easy to find relevant code
Changes localized to specific modules
Fail-Fast Principle
Principle: Validate early and fail immediately on invalid input.
Implementation:
# Configuration validation
@dataclass(frozen=True)
class HttpClientConfig:
default_timeout: float = 30.0
max_redirects: int = 10
def __post_init__(self) -> None:
# Fail fast on invalid configuration
if self.default_timeout <= 0:
raise ValueError('default_timeout must be positive')
if self.max_redirects < 0:
raise ValueError('max_redirects must be non-negative')
# Token validation
class TokenData:
def __init__(self, access_token: str, ...):
if not access_token:
raise ValueError('access_token is required')
self.access_token = access_token
Benefits:
Catch errors early
Clear error messages
Easier debugging
Explicit over Implicit
Principle: Make behavior explicit rather than relying on magic or hidden behavior.
Implementation:
# Good ✅ - Explicit
config = (
HttpClientConfigBuilder()
.with_base_url('https://api.example.com')
.with_retry(max_retries=3)
.with_timeout(30.0)
.with_bearer_token('token')
.build()
)
# Avoid ❌ - Implicit/magic
config = HttpClientConfig.from_env() # Reads from environment variables
# Where does base_url come from? What env vars? Not clear!
Explicit Error Handling:
# Explicit exception hierarchy
try:
response = client.get('/users')
except TimeoutException: # Explicit timeout error
handle_timeout()
except ConnectionException: # Explicit connection error
handle_connection_error()
Benefits:
Code is self-documenting
Easy to understand
No hidden surprises
Composition over Inheritance
Principle: Prefer object composition to class inheritance.
Implementation:
# Good ✅ - Composition
class HttpClient:
def __init__(self, config):
self._config = config
self._retry_strategy = config.retry_strategy # Composed
self._auth_hook = config.auth_hook # Composed
self._request_hooks = config.request_hooks # Composed
# Avoid ❌ - Deep inheritance
class BaseClient: ...
class RetryableClient(BaseClient): ...
class AuthenticatedClient(RetryableClient): ...
class LoggingClient(AuthenticatedClient): ...
Benefits:
More flexible
Easier to change behavior at runtime
Avoids fragile base class problem
Type Safety
Principle: Use type hints for better IDE support and runtime safety.
Implementation:
# Full type annotations
def get(
self,
url: str,
params: dict[str, Any] | None = None,
headers: dict[str, str] | None = None,
timeout: float | None = None,
**kwargs,
) -> HttpResponse:
"""Execute a GET request."""
...
# Generic types
T = TypeVar('T')
def request_many(
self,
requests_list: list[HttpRequest],
max_workers: int = 5,
fail_fast: bool = False,
) -> Generator[tuple[int, HttpResponse | HttpClientException], None, None]:
"""Execute multiple requests concurrently."""
...
Benefits:
IDE autocomplete
Type checking with mypy
Self-documenting code
Catch errors before runtime
See Also
Components Overview - System components overview
Thread Safety - Thread-safety design
Code Style Guide - Code style guide