Skip to content

Codefied-CodePix/px-sdk-py

Folders and files

NameName
Last commit message
Last commit date

Latest commit

 

History

5 Commits
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 

Repository files navigation

pxcontrol

PyPI version PyPI downloads Python versions Typed GitHub stars GitHub issues License: MIT

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.
  • 503 responses 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.1 and earlier are deprecated. The control-plane API reports 0.1.2 as the minimum supported release — older clients still run but the boot-time version check will log a warning until you upgrade.


Install

pip install pxcontrol

Set your project token:

export PX_TOKEN=px_xxxxxxxxxx

FastAPI / Starlette (ASGI)

from 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"}

Flask (WSGI)

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}

Django

# settings.py
MIDDLEWARE = [
    "pxcontrol.DjangoPxControlMiddleware",
    # ... your other middleware
]

PXCONTROL = {
    "TOKEN": "px_xxxxxxxxxx",       # optional, else PX_TOKEN env
    "HEALTH_PATH": "/healthz",       # optional
}

Plain scripts / workers

from pxcontrol import get_client

px = get_client()
if px.is_active():
    do_some_work()
else:
    print("paused/offline:", px.get_pause_message())

Behaviour

When the PXControl dashboard sets the project to:

  • ACTIVE — requests pass through.
  • PAUSED — requests return 503 with the configured pause message.
  • OFFLINE / FAILED — requests return 503 with the configured offline message.
  • The path /health (configurable) is always allowed through so external load balancers, k8s probes, and uptime monitors still work.

Fail modes

  • 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.

Configuration

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

HTTP poll fallback

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.

Version check

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.

Status listeners

from pxcontrol import get_client

px = get_client()

def on_status(new, old):
    print(f"[pxcontrol] {old} -> {new}")

px.on_status(on_status)

Graceful shutdown

from pxcontrol import reset_client

def shutdown():
    reset_client()  # closes the WebSocket, cancels background tasks

Requirements

  • Python 3.9+
  • websockets >= 13

Links

Contributing

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 + wheel

License

MIT © PXControl

About

No description, website, or topics provided.

Resources

License

Stars

Watchers

Forks

Releases

No releases published

Packages

 
 
 

Contributors

Languages