Utilities

BaseProxy

class auto_proxy_vpn.utils.base_proxy.BaseProxy[source]

Bases: ABC

ip: str

name: str

port: int

user: str

password: str

output: bool

log: bool

destroy: bool

active: bool

is_async: bool

get_proxy_str() → str[source]: Returns the url of the proxy. Empty string if no public IP yet.

get_proxy() → dict[str, str] | None[source]: Returns the proxy in a format that can be used by the requests library. Empty dict if no public IP yet.

is_active(wait: bool = False) → bool[source]

Checks if proxy is active.

Parameters:: wait (bool) – Always wait until the proxy is active, even if it is asynchronous.

close(wait: bool = True)[source]

Closes the proxy, destroying it if on_exit is set to ‘destroy’ or keeping it if on_exit is set to ‘keep’.

Parameters:: wait (bool) – Whether to wait until the proxy is fully removed when closing it. If the proxy is not asynchronous and wait is False, it will return immediately without waiting for the proxy to be fully removed. For asynchronous proxies, this parameter has no effect and the method will never wait for the proxy to be fully removed before returning.

ProxyBatch

class auto_proxy_vpn.utils.base_proxy.ProxyBatch(proxies: list[T])[source]

Bases: Generic[T]

__init__(proxies: list[T])[source]

Container for a group of proxies with iteration and lifecycle control.

The batch shuffles the incoming proxies on creation to avoid predictable ordering. It behaves as an iterable and iterator, supports indexing, and can be used as a context manager to ensure all proxies are closed when the batch is no longer needed.

Parameters:: proxies (list[BaseProxy]) – Proxies included in the batch.

Notes

Once closed, most operations raise RuntimeError.
Make sure to call close() when done to release resources if not using a context manager.

Examples

Context manager usage:

>>> with pool.create_batch(3) as batch:
...     for proxy in batch:
...         print(proxy)

Manual close:

>>> batch = pool.create_batch(3)
>>> for proxy in batch:
...     print(proxy)
>>> batch.close()

close(wait: bool = False)[source]

Close every proxy in the batch and mark the batch as closed.

Parameters:: wait (bool) – If True, wait for each proxy to close synchronously. If False, close asynchronously.

BaseProxyManager

class auto_proxy_vpn.utils.base_proxy.BaseProxyManager[source]

Bases: ABC, Generic[T]

abstractmethod classmethod from_config(config=None, runtime_config: ManagerRuntimeConfig | None = None) → BaseProxyManager[source]

abstractmethod get_proxy(port: int = 0, size: Literal['small', 'medium', 'large'] = 'medium', region: 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') → T[source]

abstractmethod get_proxy_by_name(name: str, is_async: bool = False, on_exit: Literal['destroy', 'keep'] = 'destroy') → T[source]

get_proxies(number: int, ports: list[int] | int = 0, sizes: list[Literal['small', 'medium', 'large']] | Literal['small', 'medium', 'large'] = 'medium', regions: 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[T][source]

Gets multiple proxies at once. The parameters can be a single value or a list of values. If a single value is provided, it will be used for all proxies. If a list of values is provided, it must have the same length as the number of proxies to create.

Parameters:

number (int) – Number of proxies to get.
ports (list[int] or int) – Port or list of ports to use.
sizes (list or str) – Size or list of sizes to use.
regions (list[str] or str) – Region or list of regions to use.
auths (list[dict] or dict) – Auth or list of auths to use.
allowed_ips (list[str] or str) – Allowed IP or list of allowed IPs to use in all proxies.
is_async (bool) – Whether the proxies should be async or not.
retry (bool) – Whether to retry if a proxy fails to start.
proxy_names (list[str] or str) – Proxy name or list of proxy names to use.
on_exit ({'keep', 'destroy'}) – Whether to keep or destroy the proxies when the program ends.

Returns:

Batch of proxy instances. This is the return type for get_proxies(...) in every provider manager implementation.

Return type:

ProxyBatch

Raises:

ValueError – If the length of the ports, sizes, regions, auths or proxy_names lists is not equal to the number of proxies to create, or if any of the sizes or regions is invalid.
TypeError – If any of the auths is not a dict, or if allowed_ips is not a list of strings or a string.
KeyError – If any of the auth dicts does not have the keys ‘user’ and ‘password’.

get_sizes_and_regions() → dict[Literal['small', 'medium', 'large'], list[str] | list[tuple[str, list[str]]]][source]: Get all sizes and regions avaliable.

get_regions_by_size(size: Literal['small', 'medium', 'large']) → list[str] | list[tuple[str, list[str]]][source]: Returns a list of regions for a sepecific size

abstractmethod get_running_proxy_names() → list[str] | list[tuple[str, str]][source]: Get list of running proxy names

Exceptions

exception auto_proxy_vpn.utils.exceptions.CountryNotAvailableException(*args)[source]: Bases: Exception

exception auto_proxy_vpn.utils.exceptions.ProxyIpNotAvailableException(*args)[source]: Bases: Exception

SSHClient

class auto_proxy_vpn.utils.ssh_client.SSHClient(ip: str, user: str, strict: bool = False)[source]

Bases: object

__init__(ip: str, user: str, strict: bool = False)[source]

Creates an SSH client to connect to a remote server and execute commands or download files. Checks the connection to the server before executing any command to ensure it is active.

Parameters:

ip (str) – The IP address of the remote server.
user (str) – The username to use for the SSH connection.
strict (bool) – Whether to use strict host key checking. If False, it will not check the host key and will automatically add it to the known hosts. Defaults to False.

connect()[source]: Checks server ssh connection

run_command(command: str) → tuple[int, str, str][source]

Executes a command in remote server

Parameters:: command (str) – Command to execute.
Returns:: Return the returncode, stdout and stderr of the command execution.
Return type:: tuple[int, str, str]
Raises:: ConnectionError – If the connection to the server is not active.

download_file(file: str, destination_file: str)[source]

Downloads a file from the remote server to the local machine.

Parameters:

file (str) – The path to the file on the remote server.
destination_file (str) – The path to save the file on the local machine.

Raises:

ConnectionError – If the connection to the server is not active.
FileNotFoundError – If the file does not exist on the remote server.

Utility Functions

auto_proxy_vpn.utils.util.get_public_ip(timeout=2, proxy: dict[str, str] | None = None)[source]: Get the public IP address of the machine by querying multiple external services.

auto_proxy_vpn.utils.util.is_ssh_key(key: str) → bool[source]: Check if the provided string is a valid SSH public key.

auto_proxy_vpn.utils.files_utils.get_ips_str(ips_list: list[str])[source]

auto_proxy_vpn.utils.files_utils.get_ssh_keys_str(ssh_keys: list[str], user: str = '')[source]

auto_proxy_vpn.utils.files_utils.get_squid_file(port: int, user: str = '', password: str = '', allowed_ips: list[str] = [], ssh_keys: list[str] = [], os_user: str = '') → str[source]