Skip to content

API Reference

pypia_ctl

Private Internet Access (PIA) CLI Mini-SDK — :mod:pypia_ctl.

Purpose

A compact, typed wrapper around the official :mod:piactl command for Python apps that need to programmatically control PIA and route that state into HTTP/browser clients. Provides: * Strict subprocess runner with typed exceptions. * Synchronous getters and strategy connect (preferred → random → default). * Async monitor for live updates. * :mod:pydantic_settings-backed configuration (env/.env/defaults). * Non-destructive .env helpers (create/merge/print). * (Optional) adapters for Playwright/httpx/Selenium proxy setup. * (Optional) tiny plugin protocol and loader.

Design
  • Zero hidden caching; every call shells piactl.
  • No network calls beyond invoking the PIA daemon via CLI.
  • Side effects only when calling mutators (connect, disconnect, set).
Public API

Config & bootstrap: - :class:~pypia_ctl.config.PiaSettings - :func:~pypia_ctl.bootstrap.init_settings - :func:~pypia_ctl.envtools.ensure_env_file - :func:~pypia_ctl.envtools.generate_env_text Core: - :class:~pypia_ctl.core.PiaStatus - :class:~pypia_ctl.core.MonitorEvent - :func:~pypia_ctl.core.fetch_status - :func:~pypia_ctl.core.connect_with_strategy - :func:~pypia_ctl.core.disconnect_vpn - :func:~pypia_ctl.core.get_regions - :func:~pypia_ctl.core.monitor Exceptions: - :class:~pypia_ctl.exceptions.PiaCtlError - :class:~pypia_ctl.exceptions.PiaCtlNotFound - :class:~pypia_ctl.exceptions.PiaCtlInvocationFailed - :class:~pypia_ctl.exceptions.PiaConnectTimeout Optional: - :mod:pypia_ctl.adapters (proxy helpers) - :class:~pypia_ctl.plugins.Plugin, :func:~pypia_ctl.plugins.load_plugins

Examples:

:: >>> from pypia_ctl import init_settings, fetch_status >>> s = init_settings(create_env=False) # no writes; OS env > .env > defaults >>> isinstance(s.protocol, str) True

Classes

MonitorEvent

Bases: BaseModel

One update line from piactl monitor <key>.

Parameters:

Name Type Description Default
key

The key monitored (e.g., "connectionstate").

 required
value

The text value emitted by the daemon.

 required

Returns:

Name Type Description
A

class:MonitorEvent.

Examples:

:: >>> MonitorEvent(key="connectionstate", value="Connected").value 'Connected'

PiaConnectTimeout

Bases: PiaCtlError

Did not reach Connected within the polling window.

PiaCtlError

Bases: Exception

Base error for :mod:pypia_ctl operations.

PiaCtlInvocationFailed

Bases: PiaCtlError

piactl returned non-zero or the subprocess/timeout failed.

PiaCtlNotFound

Bases: PiaCtlError

piactl executable not found on PATH.

PiaSettings

Bases: BaseSettings

User preferences & runtime knobs.

Parameters:

Name Type Description Default
protocol

"wireguard" (default) or "openvpn".

 required
default_region

Fallback region slug, e.g. "auto".

 required
preferred_regions

JSON array of region slugs; first available wins in "preferred".

 required
randomize_region

If True, allow random choice among eligible regions.

 required
subprocess_timeout_sec

Timeout (s) for individual piactl calls.

 required
monitor_line_timeout_sec

Timeout (s) per line for monitor.

 required
plugins

JSON array of fully-qualified plugin class paths.

 required
region_filters

Rules to constrain region selection.

 required
proxy

Proxy configuration for adapters.

 required
Environment
  • PIA_PROTOCOL (wireguard|openvpn)
  • PIA_DEFAULT_REGION (e.g. auto)
  • PIA_PREFERRED_REGIONS (JSON array)
  • PIA_RANDOMIZE_REGION (true/false)
  • PIA_SUBPROCESS_TIMEOUT_SEC (int)
  • PIA_MONITOR_LINE_TIMEOUT_SEC (int)
  • PIA_PLUGINS (JSON array)
  • PIA_REGION_FILTERS__include_streaming (bool)
  • PIA_REGION_FILTERS__include_countries (JSON array)
  • PIA_REGION_FILTERS__exclude_countries (JSON array)
  • PIA_PROXY__KIND (socks5|http)
  • PIA_PROXY__HOST, PIA_PROXY__PORT (int)
  • PIA_PROXY__USERNAME, PIA_PROXY__PASSWORD

PiaStatus

Bases: BaseModel

Snapshot of PIA state (JSON-log friendly).

Parameters:

Name Type Description Default
connection_state

Current state.

 required
region

Selected region slug (or "auto").

 required
regions

All known region slugs (plus "auto").

 required
vpn_ip

Current VPN IP string or None.

 required
protocol

Selected protocol (if exposed by this build).

 required
debug_logging

Whether debug logging is enabled.

 required
port_forward

Port forward model.

 required
request_port_forward

Whether PF is requested (if exposed).

 required

Returns:

Type Description

A validated :class:PiaStatus.

Examples:

:: >>> PiaStatus( ... connection_state="Disconnected", ... region="auto", ... regions=["auto"], ... vpn_ip=None, ... protocol=None, ... debug_logging=False, ... port_forward=PortForward(), ... request_port_forward=None, ... ).connection_state 'Disconnected'

Plugin

Base class for plugins.

Notes

Subclass this, provide your behavior in init/methods, and ensure the class can be instantiated without required parameters.

Functions

connect_with_strategy 

connect_with_strategy(*, strategy='preferred', exact_region=None, max_retries=2)

Connect using settings + strategy, with simple retries.

Parameters:

Name Type Description Default
strategy str

One of "preferred", "random", "exact".

 'preferred'
exact_region str | None

When strategy="exact", this slug is used.

 None
max_retries int

Reconnect attempts on failure.

 2

Raises:

Type Description
PiaCtlNotFound

piactl not on PATH.
PiaCtlInvocationFailed

piactl connect/disconnect failed.
PiaConnectTimeout

Did not reach Connected within the poll window.

Examples:

:: >>> isinstance(1, int) True

disconnect_vpn 

disconnect_vpn()

Disconnect the VPN (best effort).

Raises:

Type Description
PiaCtlNotFound

piactl not on PATH.
PiaCtlInvocationFailed

piactl disconnect failed.

Examples:

:: >>> isinstance(True, bool) True

ensure_env_file 

ensure_env_file(path, defaults=None)

Create or merge a .env file without overwriting existing keys.

Behavior
  • If the file does not exist, write defaults (plus trailing newline).
  • If it exists, append only missing keys (keep current values/comments).
  • Ignores comment/blank lines on read.

Parameters:

Name Type Description Default
path str | Path

Destination path for the .env file.

 required
defaults Iterable[str] | None

Optional custom defaults (lines like KEY=VALUE). If omitted, :data:DEFAULT_ENV is used.

 None

Returns:

Type Description
None

None

Raises:

Type Description
ValueError

If any provided default line lacks an =.

Examples:

:: >>> from pathlib import Path >>> p = Path("._ensure.env") >>> try: ... ensure_env_file(p) ... ensure_env_file(p) # idempotent ... "PIA_PROTOCOL=" in p.read_text() ... finally: ... p.unlink(missing_ok=True) True

fetch_status 

fetch_status()

Return a one-shot status snapshot by querying multiple getters.

Returns:

Name Type Description
A PiaStatus

class:PiaStatus ready for logs/health checks.

Raises:

Type Description
PiaCtlError

Any underlying runner error will bubble up.

Examples:

:: >>> isinstance(True, bool) True

generate_env_text 

generate_env_text(extra=None)

Return newline-terminated .env content.

Parameters:

Name Type Description Default
extra Mapping[str, str] | None

Optional mapping of additional key→value pairs to append.

 None

Returns:

Name Type Description
str str

Content suitable to be written to disk.

Examples:

:: >>> txt = generate_env_text({"FOO": "bar"}) >>> "FOO=bar" in txt True

get_regions 

get_regions()

Return all available region slugs (plus 'auto').

init_settings 

init_settings(create_env=False, env_path='.env', overrides=None)

Initialize :class:~pypia_ctl.config.PiaSettings.

Parameters:

Name Type Description Default
create_env bool

If True, ensure a .env exists (non-destructive merge).

 False
env_path str | None

Path to the .env to create/merge (when create_env).

 '.env'
overrides dict[str, Any] | None

In-memory overrides to apply after loading from env/.env.

 None

Returns:

Name Type Description
PiaSettings PiaSettings

A loaded, optionally overridden settings object.

Examples:

:: >>> from pypia_ctl.bootstrap import init_settings >>> s = init_settings(create_env=False, overrides={"protocol": "openvpn"}) >>> s.protocol 'openvpn'

load_plugins 

load_plugins(paths=None, *, ignore_errors=False)

Load and instantiate plugins.

Parameters:

Name Type Description Default
paths Iterable[str] | None

Iterable of "pkg.module:ClassName" entries. If None, discovery uses :class:~pypia_ctl.config.PiaSettings.plugins.

 None
ignore_errors bool

If True, skip invalid entries; otherwise raise.

 False

Returns:

Type Description
List[Plugin]

list[Plugin]: Instantiated plugins (possibly empty).

Raises:

Type Description
(ValueError, ImportError, TypeError)

When an entry is invalid (and

Examples:

:: >>> load_plugins(paths=[], ignore_errors=True) []

monitor async 

monitor(key)

Yield :class:MonitorEvent for live updates via piactl monitor <key>.

Parameters:

Name Type Description Default
key str

e.g., "connectionstate" or "vpnip".

 required

Yields:

Type Description
Iterable[MonitorEvent]

class:MonitorEvent values as lines arrive.

Raises:

Type Description
PiaCtlNotFound

If piactl is not on PATH.

Examples:

:: >>> isinstance(True, bool) True