Python SDK

Read variables that change while your process runs, and rebuild the things made from them, without a restart.

Install

pip install dynv

How it works

The same API runs in one of three modes. connect() picks one for you:

Mode	When	What happens
agent	under `dynv run` (`DYNV_AGENT_URL` is set)	The SDK talks to the small agent `dynv run` starts next to your app, over loopback, with a per-run key. The agent has already verified and decrypted the data. Your app holds no token and does no cryptography. Preferred.
direct	no `dynv run`, a service token in `DYNV_TOKEN` or `DYNV_TOKEN_FILE`	The SDK does what `dynv run` does, in your process: fetch, verify, decrypt, merge shared groups, resolve templates, watch and cache. See Without the CLI.
env	neither (local development, tests, a migration in progress)	`get()` reads `os.environ` and no change events fire, so the same code works everywhere.

Read variables

Mark the variables you want live as hot, start your app with dynv run -- python app.py, then call dynv.connect(). It waits for the first snapshot and returns a thread-safe Dynv object:

python

proxy = env.get("PROXY_SERVER")       # str | None, synchronous, always current
dsn = env.require("DATABASE_URL")     # str, or raises MissingVariableError
both = env.pick("PROXY_SERVER", "DATABASE_URL")  # one consistent snapshot

env.version                         # local revision, increasing on every applied change
env.mode                            # "agent", "direct" or "env"
env.live                            # false only in env mode
env.source("DB_HOST")               # shared group name, or None
env.is_template("DATABASE_URL")     # get() already returns the resolved value
env.unresolved()                    # template key names that do not resolve now

get() falls back to os.environ only for a key Dynv has never managed. Once Dynv has managed a key, removing it makes it read as unset, never as a stale inherited value.
A template that stops resolving keeps its last good value, or stays unset if it never had one. unresolved() returns key names only.
Read on every use. Do not save env.get(...) in a module-level constant, or you will pin the old value.

Without the CLI: direct mode

Can't put the dynv binary in your image, or can't wrap the start command? Give the app a service token instead and call connect(). The server still sends only ciphertext. The SDK verifies and decrypts it in your process, exactly as dynv run would:

it checks every signature against the signers named inside the sealed data key, and merges the shared groups attached to your service;
it watches for changes and resolves templates such as postgres://${DB_USER}@${DB_HOST}/app against your service's variables;
it keeps an encrypted cache plus the newest version it has accepted, so it starts when Dynv is down and refuses a rolled-back or forked document, even after a restart.

Set DYNV_URL explicitly in direct mode. Use https://api.dynv.dev for Dynv Cloud.

export DYNV_URL=https://api.dynv.dev
export DYNV_PROCESS=web
export DYNV_TOKEN=dynv_pt_…                      # a service token (or DYNV_TOKEN_FILE=/run/secrets/dynv)
export DYNV_CACHE=/var/lib/web/dynv-cache.json   # encrypted cache: boots when Dynv is down, refuses rollbacks

python app.py                                    # no dynv run

app.py

import threading

import dynv

# Reads DYNV_URL, DYNV_TOKEN (or DYNV_TOKEN_FILE), DYNV_PROCESS and DYNV_CACHE.
# Under `dynv run` the same call uses the local agent instead.
env = dynv.connect()
print(f"connected in {env.mode} mode, version {env.version}", flush=True)

env.require("PROXY_SERVER")  # raises if it is not set


def proxy_changed(value, change):
    # value is the new verified value, or None when the key was removed.
    print("PROXY_SERVER changed", flush=True)


env.on("PROXY_SERVER", proxy_changed)


def changed(event):
    # Key names only, never values.
    mode = "restart" if event.restart else "hot"
    print(f"changed: {', '.join(event.changed)} ({mode})", flush=True)


env.on_change(changed)

try:
    threading.Event().wait()
except KeyboardInterrupt:
    pass
finally:
    env.close()

By default, values are also mirrored into os.environ: every value at startup, then only hot changes to variables that don't control code loading (PATH, LD_PRELOAD, PYTHONPATH, …). Restart-mode and code-loading changes update get() and events but reach the environment only on the next start. Removed variables are deleted. C extensions may read a stale environment; set mirror_to_environ=False to turn mirroring off.

The tradeoff: your app holds the token

Under dynv run, DYNV_TOKEN stays with the supervisor and your app only sees a loopback key that is useless anywhere else. In direct mode your app process holds the token. Anything that can read its memory or environment can take it: a compromised dependency, an error reporter that captures env, a heap dump, a child process. A stolen token keeps reading this service's variables, including future values, until you revoke it.

What doesn't change: the server still sees only ciphertext, and a service token is still read-only and opens only its own service and its attached shared groups. To limit the damage:

the SDK deletes DYNV_TOKEN from os.environ as soon as it has read it, even if the token is then refused (scrub_token_from_env, on by default);
prefer DYNV_TOKEN_FILE pointing at a mounted secret;
use one token per deployment, and revoke it if in doubt;
never give an app an account (master) token. The SDK refuses one.

Prefer dynv run wherever you can install it.

Restart-mode variables

Nobody supervises your process in direct mode, so the SDK can't restart it. Every change is applied live. When a changed variable is in restart mode, the change event has restart=True. Pass on_restart="exit" to have the SDK send your process SIGTERM, so your platform restarts it with the new values.

Options

python

import os

import dynv

env = dynv.connect(
    mode="auto",                 # "agent" under dynv run, else "direct" with a token, else "env"
    token=None,                  # default: DYNV_TOKEN, or the file in DYNV_TOKEN_FILE
    process=None,                # default: DYNV_PROCESS; a token for another service is refused
    url=os.environ["DYNV_URL"],  # required in direct mode; https://api.dynv.dev for Dynv Cloud
    cache=None,                  # default: DYNV_CACHE; the same file format as dynv run --cache
    require=False,               # true: fail if Dynv is unreachable and there is no cache
    boot_timeout=10.0,           # seconds before the first fetch falls back to the cache
    poll_interval=120.0,         # seconds between polls when long_poll is false
    long_poll=True,              # false: poll instead (networks that drop idle connections)
    allow_insecure_http=None,    # default: DYNV_ALLOW_INSECURE_HTTP=1; otherwise HTTPS or loopback only
    scrub_token_from_env=True,   # delete DYNV_TOKEN from os.environ once read
    on_restart="ignore",         # "exit": SIGTERM yourself when a restart-mode variable changes
    mirror_to_environ=True,      # keep os.environ in step, for libraries that read it
    on_error=None,               # optional callback; also sees errors during the first fetch
    log=None,                    # default: logging warnings; key names and codes only, never values
    agent_url=None,              # default: DYNV_AGENT_URL
    agent_key=None,              # default: DYNV_AGENT_KEY
    watch=True,                  # false: load the first snapshot without starting the watcher
)
print(f"connected in {env.mode} mode", flush=True)
env.close()

React to changes

python

def on_proxy(value, change):
    # value is None if PROXY_SERVER was removed; change.key names the key.
    rebuild_proxy()

off = env.on("PROXY_SERVER", on_proxy)
off_changes = env.on_change(lambda event: print(event.version, event.changed))
off_errors = env.on_error(lambda err: print("dynv:", err))

# later: off(); off_changes(); off_errors(); env.close()

The first snapshot is the starting point, not a change. Synchronous callbacks run on the watcher thread; keep them short and hand off blocking work. An async def callback must be registered from inside a running event loop, and runs on that loop. Errors and refused updates go to on_error; the SDK keeps the last good values and retries.

For asyncio, aconnect() moves the blocking first fetch to a worker thread. changes() is an async iterator over the same change events:

app.py

import asyncio

import dynv


async def main():
    env = await dynv.aconnect()
    print(f"connected in {env.mode} mode, version {env.version}", flush=True)
    env.require("PROXY_SERVER")

    proxy_changed = asyncio.Event()

    async def on_proxy(value, change):
        # Registered inside the running loop, so this coroutine runs on it too.
        print("PROXY_SERVER changed", flush=True)
        proxy_changed.set()

    env.on("PROXY_SERVER", on_proxy)
    changes = env.changes()
    try:
        async for event in changes:
            print(f"changed: {', '.join(event.changed)}", flush=True)
            if "PROXY_SERVER" in event.changed:
                await proxy_changed.wait()
                break
    finally:
        await changes.aclose()
        env.close()


asyncio.run(main())

Inspect an environment from the command line

The package includes a small helper for direct and agent mode. python -m dynv env prints one verified snapshot as JSON. python -m dynv watch prints the first snapshot and then one JSON line per change. Both use the same DYNV_* environment variables as connect().

Using another runtime? See the Node / TypeScript SDK, the Go SDK, or the protocol and test vectors.