Core

CloudProvider

class auto_proxy_vpn.CloudProvider(*values)

Bases: str, Enum

Enum for supported cloud providers. Only providers defined here will be registered to use in ProxyPool. The value should match the name of the provider package in auto_proxy_vpn.providers

GOOGLE = 'google'

AZURE = 'azure'

DIGITALOCEAN = 'digitalocean'

AWS = 'aws'

ProxyManagers

class auto_proxy_vpn.manager_register.ProxyManagers

Bases: ABC

Central registry for provider-specific proxy manager classes.

This class stores a mapping between CloudProvider values and the corresponding manager class that handles proxy creation for that provider. Manager implementations register themselves through the register() decorator.

Notes

• Registration is global for the current Python process.

• A provider can only be registered once.

classmethod register(provider: CloudProvider)

Class decorator to register a proxy manager class for a specific cloud provider.

classmethod get_manager(provider: CloudProvider) → type[BaseProxyManager]

Return the registered manager class for a cloud provider.

Parameters:

provider (CloudProvider) – Provider whose manager class is requested.

Returns:

Registered manager class for provider.

Return type:

type[BaseProxyManager]

Raises:

ValueError – If no manager class is registered for provider.

ProxyPool

class auto_proxy_vpn.proxy_pool.ProxyPool(*provider_configs: BaseConfig, log: bool = True, log_file: str | None = None, log_format: str = '%(asctime)-10s %(levelname)-5s %(message)s', logger: Logger | None = None)

Bases: object

High-level orchestrator for creating proxies across cloud providers.

ProxyPool initializes one manager per selected provider and exposes convenience methods to create a single proxy or a batch of proxies. When multiple providers are configured, manager selection is randomized per cycle using RandomManagerPicker.

Notes

• Duplicate providers are not allowed.

• At most one configuration per provider is accepted.

• Manager-specific validation is delegated to each provider manager.

__init__(*provider_configs: BaseConfig, log: bool = True, log_file: str | None = None, log_format: str = '%(asctime)-10s %(levelname)-5s %(message)s', logger: Logger | None = None)

Build a proxy pool and initialize provider managers. Supports multiple accounts per provider by passing multiple configurations. To be able to use multiple managers for the same provider the credentials must be different, otherwise an error will be raised since the managers would be indistinguishable.

Parameters:

• *provider_configs (tuple[BaseConfig, ...]) –

  One or more provider-specific configuration objects.

  Example:

  # will create 4 managers (2 google managers with different accounts)
  ProxyPool(GoogleConfig(...), AzureConfig(...), GoogleConfig(credentials='credentials_2.json', ...), DigitalOceanConfig(...))
  

• log (bool, optional) – Enable logging for manager runtime. Defaults to True.

• log_file (str | None, optional) – Path to a log file. If None, logging goes to the default handler.

• log_format (str, optional) – Format string used when configuring logging.

• logger (Logger | None, optional) – Shared logger instance passed to all managers.

Raises:

ValueError – If no providers are supplied, if providers contain duplicates, or if provider configurations are duplicated by provider.

Examples

Single provider with config:

from auto_proxy_vpn import ProxyPool, GoogleConfig
google_config = GoogleConfig(project='my-google-project', ssh_key='ssh_keys')
pool = ProxyPool(google_config)
batch = pool.create_batch(3)
for proxy in batch:
    print(proxy)
batch.close()

Multiple providers with shared logger:

from auto_proxy_vpn import ProxyPool, CloudProvider, GoogleConfig, AzureConfig
google_config = GoogleConfig(project='my-google-project', ssh_key='ssh_keys')
azure_config = AzureConfig(ssh_key='ssh_keys')
pool = ProxyPool(google_config, azure_config)
with pool.create_one() as proxy:
    # do something with the proxy
    pass

create_one(port: int = 0, size: Literal['small', 'medium', 'large'] = 'medium', region: dict[CloudProvider, str] = {}, auth: dict[Literal['user', 'password'], str] = {}, allowed_ips: str | list[str] = [], is_async: bool = False, retry: bool = True, proxy_name: str = '', on_exit: Literal['keep', 'destroy'] = 'destroy') → BaseProxy

Create a single proxy using the next randomly selected manager.

Parameters:

• port (int, optional) – Desired proxy port. Defaults to 0 (random).

• size (Literal['small', 'medium', 'large'], optional) – Desired proxy size. Defaults to 'medium'.

• region (str, optional) – Desired proxy region. Defaults to '' (any).

• auth (dict[Literal['user', 'password'], str], optional) – Authentication credentials. Defaults to empty dict.

• allowed_ips (str | list[str], optional) – Allowed IPs for proxy access. Defaults to empty list.

• is_async (bool, optional) – Whether to create the proxy asynchronously. Defaults to False.

• retry (bool, optional) – Whether to retry on failure. Defaults to True.

• proxy_name (str, optional) – Optional name for the proxy. Defaults to empty string.

• on_exit (Literal['keep', 'destroy'], optional) – Action to take when the proxy is closed. Defaults to 'destroy'.

Returns:

The created proxy instance.

Return type:

BaseProxy

create_batch(count: int, ports: list[int] | int = 0, sizes: list[Literal['small', 'medium', 'large']] | Literal['small', 'medium', 'large'] = 'medium', regions: dict[CloudProvider, list[str] | str] = {}, auths: list[dict[Literal['user', 'password'], str]] | dict[Literal['user', 'password'], str] = {}, allowed_ips: list[str] | str = [], is_async: bool = True, retry: bool = True, proxy_names: list[str] | str = '', on_exit: Literal['keep', 'destroy'] = 'destroy') → ProxyBatch[BaseProxy]

Create multiple proxies distributed as evenly as possible.

The requested count is split across configured providers using an even distribution with remainder. For each non-zero share, one manager is selected from RandomManagerPicker and asked to create its assigned number of proxies.

Parameters are forwarded to each manager’s get_proxies method.

Returns:

Batch containing all successfully created proxies.

Return type:

ProxyBatch

Notes

If a provider manager handles internal failures by returning fewer proxies, the resulting batch may contain fewer proxies than requested.

get_sizes_and_regions(provider: CloudProvider) → dict[Literal['small', 'medium', 'large'], list[str] | list[tuple[str, list[str]]]]

Get available regions for a specific provider.

Parameters:

provider (CloudProvider) – The cloud provider for which to retrieve regions.

Returns:

A dictionary mapping proxy sizes to their available regions.

Return type:

dict[Literal[‘small’, ‘medium’, ‘large’], list[str] | list[tuple[str, list[str]]]]

Raises:

ValueError – If the specified provider is not configured in the pool.

get_regions_by_size(provider: CloudProvider, size: Literal['small', 'medium', 'large']) → list[str] | list[tuple[str, list[str]]]

Get available regions for a specific provider and proxy size.

Parameters:

• provider (CloudProvider) – The cloud provider for which to retrieve regions.

• size (Literal['small', 'medium', 'large']) – The proxy size for which to retrieve regions.

Returns:

A list of regions where the specified proxy size is available.

Return type:

list[str] | list[tuple[str, list[str]]]

Raises:

ValueError – If the specified provider is not configured in the pool or if the specified size is invalid for that provider.

get_running_proxy_names() → dict[CloudProvider, dict[str, list[str] | list[tuple[str, str]]]]

Get names of currently running proxies grouped by provider and account.

This method supports multiple accounts for a single provider. For each provider, entries are grouped by account order using keys account_1, account_2, etc. Account numbering preserves the same order used when configs were passed to ProxyPool during initialization.

Returns:

Nested mapping where each provider contains one dictionary per configured account/manager, and each account value is the exact output returned by that manager’s get_running_proxy_names().

Return type:

dict[CloudProvider, dict[str, list[str] | list[tuple[str, str]]]]

Examples

Example output:

{
    CloudProvider.GOOGLE: {
        "account_1": ["proxy1", "proxy2"],
        "account_2": ["proxy1"],
    },
    CloudProvider.AWS: {
        "account_1": [("proxy1", "proxy2")],
    },
}

RandomManagerPicker

class auto_proxy_vpn.proxy_pool.RandomManagerPicker(managers: list[tuple[CloudProvider, BaseProxyManager]])

Bases: object

__init__(managers: list[tuple[CloudProvider, BaseProxyManager]])

Round-robin random picker for proxy managers.

The picker keeps a shuffled “bag” of manager instances and returns one manager at a time via next(). Once the bag is empty, it is refilled and shuffled again. This guarantees each manager is selected at most once per cycle while still randomizing the order between cycles.

Parameters:

managers (list[tuple[CloudProvider, BaseProxyManager]]) – List of manager instances.

Notes

• Selection is random per cycle, not weighted.

• If only one manager is provided, it is always returned.

next() → tuple[CloudProvider, BaseProxyManager]

Return the next manager, refilling the bag when needed.