proxywhirl.models.core

Data models for ProxyWhirl using Pydantic v2.

Classes

BootstrapConfig	Configuration for the lazy bootstrap that runs on first request when the pool is empty.
CircuitBreakerConfig	Configuration for circuit breaker behavior.
HealthMonitor	Continuous health monitoring for proxy pools with auto-eviction.
HealthStatus	Proxy health status states.
PoolSummary	Summary of entire proxy pool.
Proxy	Represents a single proxy server with connection details and metadata.
ProxyChain	Represents a chain of proxies for tunneling requests.
ProxyConfiguration	Global configuration settings for the proxy system.
ProxyCredentials	Secure credential storage for proxy authentication.
ProxyFormat	Supported proxy list formats.
ProxyPool	Collection of proxies with management capabilities.
ProxySource	Origin type of proxy.
ProxySourceConfig	Configuration for a proxy list source.
ProxyStatus	Status of a single proxy in the pool.
RenderMode	Page rendering modes for fetching proxy lists.
RequestResult	Result of HTTP request made through proxy.
SelectionContext	Context information for proxy selection decisions.
Session	Session tracking for sticky proxy assignments.
SourceStats	Statistics for a proxy source.
StorageBackend	Protocol for proxy storage backends.
StrategyConfig	Configuration for rotation strategies.
ValidationLevel	Proxy validation strictness levels.

Module Contents

class proxywhirl.models.core.BootstrapConfig(/, **data)

Bases: pydantic.BaseModel

Configuration for the lazy bootstrap that runs on first request when the pool is empty.

Parameters:

data (Any)

enabled

Whether automatic bootstrap is enabled (set False to supply proxies manually)

sources

Explicit list of sources to fetch from. None = random sample from ALL_SOURCES.

sample_size

How many sources to randomly pick when sources is None

validate_proxies

Whether to validate fetched proxies

timeout

Validation timeout in seconds

max_concurrent

Maximum concurrent validation requests

show_progress

Show Rich progress bars. None = auto-detect TTY on stderr.

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

model_config

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class proxywhirl.models.core.CircuitBreakerConfig(/, **data)

Bases: pydantic.BaseModel

Configuration for circuit breaker behavior.

Parameters:

data (Any)

failure_threshold

Number of failures before opening circuit

window_duration

Rolling window duration in seconds

timeout_duration

How long circuit stays open before testing recovery

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

model_config

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class proxywhirl.models.core.HealthMonitor(pool, check_interval=60, failure_threshold=3)

Continuous health monitoring for proxy pools with auto-eviction.

Runs background health checks at configurable intervals and automatically evicts proxies that fail consecutive checks beyond a threshold.

Example

>>> pool = ProxyPool(name="my_pool")
>>> pool.add_proxy(Proxy(url="http://proxy.com:8080"))
>>> monitor = HealthMonitor(pool=pool, check_interval=60, failure_threshold=3)
>>> await monitor.start()  # Start background checks
>>> # ... proxies are monitored continuously ...
>>> await monitor.stop()  # Stop monitoring

Initialize health monitor.

Parameters:

• pool (ProxyPool) – ProxyPool to monitor

• check_interval (int) – Seconds between health checks (default: 60)

• failure_threshold (int) – Consecutive failures before eviction (default: 3)

Raises:

ValueError – If check_interval or failure_threshold <= 0

get_status()

Get current monitoring status.

Returns:

Monitoring status including is_running, check_interval, failure_threshold, total_proxies, healthy_proxies, failure_counts, and uptime_seconds (if running).

Return type:

dict[str, Any]

async start()

Start background health monitoring.

Creates an asyncio task that runs health checks periodically. Idempotent - calling multiple times won’t create duplicate tasks.

Return type:

None

async stop()

Stop background health monitoring.

Cancels the background task and waits for graceful shutdown. Idempotent - safe to call when not running.

Return type:

None

class proxywhirl.models.core.HealthStatus

Bases: str, enum.Enum

Proxy health status states.

Initialize self. See help(type(self)) for accurate signature.

class proxywhirl.models.core.PoolSummary(/, **data)

Bases: pydantic.BaseModel

Summary of entire proxy pool.

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

Parameters:

data (Any)

class proxywhirl.models.core.Proxy(/, **data)

Bases: pydantic.BaseModel

Represents a single proxy server with connection details and metadata.

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

Parameters:

data (Any)

complete_request(success, response_time_ms, alpha=None)

Mark a request as complete and update metrics.

Parameters:

• success (bool) – Whether the request was successful

• response_time_ms (float) – Response time in milliseconds

• alpha (float | None) – Optional EMA smoothing factor. Strategies can pass their configured alpha to ensure consistent metric calculations without mutating proxy state.

Return type:

None

This method decrements requests_active, increments requests_completed, updates EMA response time, and delegates to record_success/record_failure.

Thread-safe: Uses internal lock to prevent race conditions.

extract_protocol_from_url()

Extract protocol from URL if not explicitly provided.

Return type:

Proxy

is_window_expired()

Check if the current sliding window has expired.

Returns:

True if window duration has elapsed since window_start

Return type:

bool

Thread-safe: Uses internal lock to prevent race conditions.

record_failure(error=None)

Record a failed request.

Parameters:

error (str | None)

Return type:

None

record_success(response_time_ms, alpha=None)

Record a successful request.

Parameters:

• response_time_ms (float) – Response time in milliseconds

• alpha (float | None) – Optional EMA smoothing factor for metrics update

Return type:

None

reset_if_expired()

Atomically check if window is expired and reset if needed.

This method prevents Time-of-Check to Time-of-Use (TOCTOU) race conditions by performing the expiration check and reset within a single lock acquisition.

Returns:

True if window was expired and has been reset, False otherwise

Return type:

bool

Thread-safe: Uses internal lock to ensure atomic check-and-reset.

reset_window()

Reset the sliding window counters.

This is called when the window duration has elapsed, preventing counter staleness and unbounded memory growth.

Thread-safe: Uses internal lock to prevent race conditions.

Return type:

None

set_expiration_from_ttl()

Set expires_at timestamp if ttl is provided.

Return type:

Proxy

start_request()

Mark a request as started (for tracking in-flight requests).

This should be called when a request is about to be made through this proxy. Increments both requests_started and requests_active counters.

Thread-safe: Uses internal lock to prevent race conditions.

Return type:

None

update_metrics(response_time_ms, alpha=None)

Update EMA and average response time metrics.

This method centralizes all response time metric updates to ensure consistency across the codebase. Both average_response_time_ms and ema_response_time_ms use the same alpha value.

Parameters:

• response_time_ms (float) – Response time in milliseconds

• alpha (float | None) – EMA smoothing factor (0-1). Higher values weight recent observations more heavily. If not provided, falls back to self.ema_alpha for backward compatibility. Strategies should pass their configured StrategyConfig.ema_alpha for consistent behavior, or use StrategyState for independent per-proxy metrics.

Return type:

None

classmethod validate_counters_non_negative(v)

Validate counter fields are non-negative.

Parameters:

v (int)

Return type:

int

validate_credentials()

Ensure username and password are both present or both absent.

Return type:

Proxy

classmethod validate_latency_non_negative(v)

Validate latency_ms is non-negative.

Parameters:

v (float | None)

Return type:

float | None

validate_local_addresses()

Reject localhost/internal IPs unless allow_local=True.

Return type:

Proxy

classmethod validate_port_range(v)

Validate port number is in valid range (1-65535).

Parameters:

v (str)

Return type:

str

classmethod validate_response_time_non_negative(v)

Validate response time fields are non-negative.

Parameters:

v (float | None)

Return type:

float | None

classmethod validate_url_scheme(v)

Validate URL has a valid proxy scheme.

Parameters:

v (str)

Return type:

str

property credentials: ProxyCredentials | None

Get credentials if present.

Return type:

ProxyCredentials | None

property is_expired: bool

Check if proxy has expired based on TTL or explicit expiration time.

Returns:

True if proxy has expired, False otherwise. Always False if no TTL/expires_at set.

Return type:

bool

property is_healthy: bool

Check if proxy is healthy.

Return type:

bool

property success_rate: float

Calculate success rate, clamped to [0.0, 1.0].

Return type:

float

class proxywhirl.models.core.ProxyChain(/, **data)

Bases: pydantic.BaseModel

Represents a chain of proxies for tunneling requests.

A proxy chain allows requests to be routed through multiple proxies in sequence, where each proxy connects to the next one in the chain. This is useful for: - Enhanced anonymity through multi-hop routing - Geographic routing through specific locations - Bypassing network restrictions

Note: Full CONNECT tunneling implementation requires additional infrastructure. This model provides the data structure and basic support for chaining.

Example

>>> chain = ProxyChain(
...     proxies=[
...         Proxy(url="http://proxy1.com:8080"),
...         Proxy(url="http://proxy2.com:8080"),
...     ],
...     name="geo_chain",
...     description="Route through US then UK"
... )

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

Parameters:

data (Any)

get_chain_urls()

Get list of proxy URLs in chain order.

Returns:

List of proxy URLs from entry to exit

Return type:

list[str]

validate_chain_length()

Ensure chain has at least 2 proxies.

Return type:

ProxyChain

property chain_length: int

Get the number of proxies in the chain.

Return type:

int

property entry_proxy: Proxy

Get the first proxy in the chain (entry point).

Return type:

Proxy

property exit_proxy: Proxy

Get the last proxy in the chain (exit point).

Return type:

Proxy

class proxywhirl.models.core.ProxyConfiguration(_case_sensitive=None, _nested_model_default_partial_update=None, _env_prefix=None, _env_file=ENV_FILE_SENTINEL, _env_file_encoding=None, _env_ignore_empty=None, _env_nested_delimiter=None, _env_nested_max_split=None, _env_parse_none_str=None, _env_parse_enums=None, _cli_prog_name=None, _cli_parse_args=None, _cli_settings_source=None, _cli_parse_none_str=None, _cli_hide_none_type=None, _cli_avoid_json=None, _cli_enforce_required=None, _cli_use_class_docs_for_groups=None, _cli_exit_on_error=None, _cli_prefix=None, _cli_flag_prefix_char=None, _cli_implicit_flags=None, _cli_ignore_unknown_args=None, _cli_kebab_case=None, _cli_shortcuts=None, _secrets_dir=None, **values)

Bases: pydantic_settings.BaseSettings

Global configuration settings for the proxy system.

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

Parameters:

• _case_sensitive (bool | None)

• _nested_model_default_partial_update (bool | None)

• _env_prefix (str | None)

• _env_file (pydantic_settings.sources.DotenvType | None)

• _env_file_encoding (str | None)

• _env_ignore_empty (bool | None)

• _env_nested_delimiter (str | None)

• _env_nested_max_split (int | None)

• _env_parse_none_str (str | None)

• _env_parse_enums (bool | None)

• _cli_prog_name (str | None)

• _cli_parse_args (bool | list[str] | tuple[str, Ellipsis] | None)

• _cli_settings_source (pydantic_settings.sources.CliSettingsSource[Any] | None)

• _cli_parse_none_str (str | None)

• _cli_hide_none_type (bool | None)

• _cli_avoid_json (bool | None)

• _cli_enforce_required (bool | None)

• _cli_use_class_docs_for_groups (bool | None)

• _cli_exit_on_error (bool | None)

• _cli_prefix (str | None)

• _cli_flag_prefix_char (str | None)

• _cli_implicit_flags (bool | None)

• _cli_ignore_unknown_args (bool | None)

• _cli_kebab_case (bool | None)

• _cli_shortcuts (collections.abc.Mapping[str, str | list[str]] | None)

• _secrets_dir (pydantic_settings.sources.PathType | None)

• values (Any)

classmethod validate_positive(v)

Ensure value is positive.

Parameters:

v (int)

Return type:

int

validate_storage()

Validate storage configuration.

Return type:

ProxyConfiguration

model_config

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class proxywhirl.models.core.ProxyCredentials(/, **data)

Bases: pydantic.BaseModel

Secure credential storage for proxy authentication.

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

Parameters:

data (Any)

to_dict(reveal=False)

Serialize credentials, optionally revealing secrets.

Parameters:

reveal (bool)

Return type:

dict[str, Any]

to_httpx_auth()

Convert to httpx BasicAuth object.

Return type:

httpx.BasicAuth

class proxywhirl.models.core.ProxyFormat

Bases: str, enum.Enum

Supported proxy list formats.

Initialize self. See help(type(self)) for accurate signature.

class proxywhirl.models.core.ProxyPool(**data)

Bases: pydantic.BaseModel

Collection of proxies with management capabilities.

Thread-safe proxy pool with RLock protection for concurrent access. All mutating operations are automatically protected by a reentrant lock.

Initialize ProxyPool with thread lock and ID index.

Parameters:

data (Any)

add_proxy(proxy)

Add proxy to pool (thread-safe).

Parameters:

proxy (Proxy)

Return type:

None

clear_expired()

Remove all expired proxies, return count removed (thread-safe).

Returns:

Number of expired proxies removed from the pool

Return type:

int

clear_unhealthy()

Remove all unhealthy proxies, return count removed (thread-safe).

Return type:

int

filter_by_source(source)

Get proxies from specific source (thread-safe).

Parameters:

source (ProxySource)

Return type:

list[Proxy]

filter_by_tags(tags)

Get proxies matching all tags (thread-safe).

Parameters:

tags (set[str])

Return type:

list[Proxy]

get_all_proxies()

Get all proxies in the pool (thread-safe).

Return type:

list[Proxy]

get_healthy_proxies()

Get all healthy or unknown (not yet tested) proxies, excluding expired ones (thread-safe).

Returns only proxies that are: - Healthy, degraded, or unknown status - Not expired (if TTL is set)

Return type:

list[Proxy]

get_proxy_by_id(proxy_id)

Find proxy by ID using O(1) index lookup (thread-safe).

Parameters:

proxy_id (uuid.UUID)

Return type:

Proxy | None

get_source_breakdown()

Get count of proxies by source (thread-safe).

Return type:

dict[str, int]

remove_proxy(proxy_id)

Remove proxy from pool (thread-safe).

Parameters:

proxy_id (uuid.UUID)

Return type:

None

property healthy_count: int

Count healthy proxies (thread-safe).

Return type:

int

model_config

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

property overall_success_rate: float

Weighted average success rate (thread-safe).

Return type:

float

property size: int

Get pool size (thread-safe).

Return type:

int

property total_requests: int

Sum of all proxy requests (thread-safe).

Return type:

int

property unhealthy_count: int

Count unhealthy proxies (thread-safe).

Return type:

int

class proxywhirl.models.core.ProxySource

Bases: str, enum.Enum

Origin type of proxy.

Initialize self. See help(type(self)) for accurate signature.

class proxywhirl.models.core.ProxySourceConfig(/, **data)

Bases: pydantic.BaseModel

Configuration for a proxy list source.

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

Parameters:

data (Any)

model_config

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class proxywhirl.models.core.ProxyStatus(/, **data)

Bases: pydantic.BaseModel

Status of a single proxy in the pool.

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

Parameters:

data (Any)

class proxywhirl.models.core.RenderMode

Bases: str, enum.Enum

Page rendering modes for fetching proxy lists.

Initialize self. See help(type(self)) for accurate signature.

class proxywhirl.models.core.RequestResult(/, **data)

Bases: pydantic.BaseModel

Result of HTTP request made through proxy.

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

Parameters:

data (Any)

is_success()

Check if request was successful.

Return type:

bool

class proxywhirl.models.core.SelectionContext(/, **data)

Bases: pydantic.BaseModel

Context information for proxy selection decisions.

This model captures all the contextual information that might be relevant for intelligent proxy selection, including session tracking, target URL characteristics, and previous attempt history.

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

Parameters:

data (Any)

model_config

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class proxywhirl.models.core.Session(/, **data)

Bases: pydantic.BaseModel

Session tracking for sticky proxy assignments.

This model maintains the relationship between a session and its assigned proxy, with TTL support and usage tracking.

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

Parameters:

data (Any)

is_expired()

Check if the session has expired.

Returns:

True if current time >= expires_at

Return type:

bool

touch()

Update last_used_at timestamp and increment request_count.

Return type:

None

model_config

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class proxywhirl.models.core.SourceStats(/, **data)

Bases: pydantic.BaseModel

Statistics for a proxy source.

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

Parameters:

data (Any)

class proxywhirl.models.core.StorageBackend

Bases: Protocol

Protocol for proxy storage backends.

This defines the interface that all storage backends must implement, allowing for file-based, database, or other storage mechanisms.

async clear()

Clear all proxies from storage.

Raises:

IOError – If clear operation fails

Return type:

None

async load()

Load proxies from storage.

Returns:

List of proxies loaded from storage

Raises:

• FileNotFoundError – If storage doesn’t exist

• ValueError – If data is corrupted or invalid

Return type:

list[Proxy]

async save(proxies)

Save proxies to storage.

Parameters:

proxies (list[Proxy]) – List of proxies to save

Raises:

IOError – If save operation fails

Return type:

None

class proxywhirl.models.core.StrategyConfig(/, **data)

Bases: pydantic.BaseModel

Configuration for rotation strategies.

This model provides flexible configuration options for all rotation strategies, allowing customization of weights, EMA parameters, session settings, and geo-targeting constraints.

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

Parameters:

data (Any)

model_config

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class proxywhirl.models.core.ValidationLevel

Bases: str, enum.Enum

Proxy validation strictness levels.

BASIC: Format + TCP connectivity validation (fast, ~100ms) STANDARD: BASIC + HTTP request test (medium, ~500ms) FULL: STANDARD + Anonymity check (comprehensive, ~2s)

Initialize self. See help(type(self)) for accurate signature.