Python SDK and middleware for PXControl — a remote application lifecycle controller. Pause, resume, fail, or take a service offline from the PXControl dashboard with no redeploy.
One line of middleware in FastAPI, Starlette, Flask, or Django and you get:
- Real-time status updates over WebSocket (plus HTTP poll fallback).
- Auto-reconnect with exponential backoff + jitter.
503responses with operator-configured pause / offline messages.- Health-check path is always whitelisted (LBs and probes keep working).
- Heartbeats so the dashboard can tell you're alive.
- Sync and async friendly — the client runs its own asyncio loop in a daemon thread, so Flask / Django / scripts get the same real-time behaviour as FastAPI.
Heads-up:
0.1.1and earlier are deprecated. The control-plane API reports0.1.2as the minimum supported release — older clients still run but the boot-time version check will log a warning until you upgrade.
pip install pxcontrolSet your project token:
export PX_TOKEN=px_xxxxxxxxxxfrom fastapi import FastAPI
from pxcontrol import PxControlASGIMiddleware
app = FastAPI()
app.add_middleware(PxControlASGIMiddleware)
@app.get("/")
def root():
return {"ok": True}
@app.get("/health")
def health():
return {"status": "ok"}from flask import Flask
from pxcontrol import PxControlWSGIMiddleware
app = Flask(__name__)
app.wsgi_app = PxControlWSGIMiddleware(app.wsgi_app)
@app.get("/")
def root():
return {"ok": True}# settings.py
MIDDLEWARE = [
"pxcontrol.DjangoPxControlMiddleware",
# ... your other middleware
]
PXCONTROL = {
"TOKEN": "px_xxxxxxxxxx", # optional, else PX_TOKEN env
"HEALTH_PATH": "/healthz", # optional
}from pxcontrol import get_client
px = get_client()
if px.is_active():
do_some_work()
else:
print("paused/offline:", px.get_pause_message())When the PXControl dashboard sets the project to:
ACTIVE— requests pass through.PAUSED— requests return503with the configured pause message.OFFLINE/FAILED— requests return503with the configured offline message.- The path
/health(configurable) is always allowed through so external load balancers, k8s probes, and uptime monitors still work.
closed(default, recommended for production) — if the SDK loses its WebSocket connection to PXControl, it assumes the service is offline and starts blocking traffic.open— if the SDK loses its connection, traffic keeps flowing until the status is explicitly changed.
Configure this in the PXControl dashboard per project.
Every argument can also be set via an environment variable.
| Argument | Env var | Default | Description |
|---|---|---|---|
token |
PX_TOKEN |
— | Project token (px_…). Required. |
ws_url |
PX_WS_URL |
wss://api-pxcontrol.codefied.online |
WebSocket base URL |
api_url |
PX_API_URL |
derived from ws_url |
HTTP base for poll fallback + version check |
heartbeat_interval |
— | 30.0 seconds |
Heartbeat cadence |
reconnect_delay |
— | 1.0 second |
Initial backoff delay (full-jitter exponential) |
max_reconnect_delay |
— | 60.0 seconds |
Upper bound for the reconnect delay |
health_path |
— | /health |
Path that always bypasses gating |
debug |
PX_DEBUG |
false |
Verbose logging |
environment |
PX_ENV / APP_ENV |
unknown |
Sent in the heartbeat payload |
disable_poll_fallback |
PX_DISABLE_POLL |
false |
Turn off the HTTP poll fallback |
poll_interval |
— | 15.0 seconds |
Poll cadence while in fallback mode |
poll_fallback_after_failures |
— | 3 |
Consecutive WS failures before polling kicks in |
disable_version_check |
PX_DISABLE_VERSION_CHECK |
false |
Skip the boot-time SDK version check |
After poll_fallback_after_failures consecutive WebSocket failures, the
client starts calling GET {api_url}/api/v1/status/{token} every
poll_interval seconds while continuing to retry the WS in the
background. Polling stops automatically once the WebSocket reconnects.
Uses urllib.request — no extra runtime dependency.
Boot-time GET {api_url}/api/v1/sdk/versions compares the installed
SDK against the server's python.latest. If older, a warning is logged
via the pxcontrol logger. The SDK never auto-updates itself.
from pxcontrol import get_client
px = get_client()
def on_status(new, old):
print(f"[pxcontrol] {old} -> {new}")
px.on_status(on_status)from pxcontrol import reset_client
def shutdown():
reset_client() # closes the WebSocket, cancels background tasks- Python 3.9+
websockets >= 13
- Source:
Codefied-CodePix/px-sdk-py - Issues: github.com/Codefied-CodePix/px-sdk-py/issues
- Changelog:
CHANGELOG.md - Dashboard: https://pxcontrol.codefied.online
- API: https://api-pxcontrol.codefied.online
- Node.js SDK:
pxcontrol-sdkon npm · source
PRs welcome. Clone and set up a dev env:
python -m venv .venv && source .venv/bin/activate
pip install -e ".[dev]"
pytest # tests
ruff check . # lint
python -m build # sdist + wheelMIT © PXControl