treinta-previews 0.2.0

Python SDK for the Propie / Previews VM-as-a-Service platform (Firecracker microVMs).

treinta-previews (Python SDK)

Python SDK for the Propie / Previews VM-as-a-Service platform — launch Firecracker microVMs from a git repo or a local folder, run commands, stream logs, attach drives and databases, take snapshots, promote to permanent (scale-to-zero) previews, and embed a browser widget safely.

It mirrors the Node SDK (cli/client.ts + web/src/lib/api.ts) 1:1 in surface and naming, translated to Pythonic snake_case.

Install

pip install treinta-previews

• Distribution name: treinta-previews
• Import name: previews

from previews import PreviewsClient

If the top-level name previews ever collides with another package in your environment, the intended fallback import name is treinta_previews. This release publishes the package as previews; use a virtualenv to avoid collisions.

Optional extras:

pip install "treinta-previews[fastapi]"   # widget proxy FastAPI adapter
pip install "treinta-previews[flask]"     # widget proxy Flask adapter

Authentication

The client resolves credentials from arguments or the environment:

Setting	Argument	Env vars (in precedence order)	Default
API key	api_key	TREINTA_PREVIEWS_API_KEY, PROPIE_API_KEY, PREVIEWS_API_KEY	—
Base URL	base_url	TREINTA_PREVIEWS_API_URL, PROPIE_API_URL	https://previews.amapola.treinta.ai/api

Keys look like pvk_<prefix>_<secret> and are sent as Authorization: Bearer pvk_....

Quick start

from previews import PreviewsClient

with PreviewsClient() as client:            # api key from env
    vm = client.vms.create(
        repo_url="https://github.com/owner/repo",
        stack="node20",
        exposed_port=3000,
        environment_variables={"NODE_ENV": "production"},
    )
    vm = client.vms.wait_until_running(vm.id)
    print(vm.url)

    result = client.vms.run(vm.id, "npm test")
    print(result.exit_code, result.stdout)

    for event in client.vms.logs(vm.id, follow=False):
        print(event.message)

    client.vms.destroy(vm.id)

Deploy a local folder (zipped locally, honoring .gitignore, stripping secrets):

det = client.detect.folder("/path/to/app")           # SSE stack detection
vm = client.vms.create_from_folder(
    "/path/to/app",
    stack=det.stack,
    start_command=det.start_command,
    exposed_port=det.exposed_port,
)

Async

from previews import AsyncPreviewsClient

async with AsyncPreviewsClient() as client:
    vms = await client.vms.list()
    async for event in client.vms.logs(vms[0].id, follow=False):
        print(event.message)

API surface

PreviewsClient(api_key=None, *, base_url=None, timeout=30.0, http_client=None) (and the identical AsyncPreviewsClient with await/async generators/aclose()). Both are context managers and expose request(method, path, *, json=None, ...) plus the resources below.

client.vms

Method	REST
create(**fields)	POST /vms (json)
create_from_folder(path, **meta) / create_from_zip(zip_bytes, **meta)	POST /vms (multipart)
list()	GET /vms
get(id)	GET /vms/:id
destroy(id)	DELETE /vms/:id
restart(id)	POST /vms/:id/restart
redeploy(id)	POST /vms/:id/redeploy
revive(id)	POST /vms/:id/revive
run(id, command, *, timeout=300.0)	POST /vms/:id/run (buffered)
logs(id, *, follow=True)	GET /vms/:id/logs (SSE) → Iterator[LogEvent]
get_env(id) → list[str] (keys only; values write-only) / set_env(id, env)	GET/PUT /vms/:id/env
get_firewall(id) / set_firewall(id, *, default_action, ip_rules, rate_limit=None)	GET/PUT /vms/:id/firewall
bandwidth(id)	GET /vms/:id/bandwidth
persist(id, slug) / unpersist(id) / rename_slug(id, slug)	/vms/:id/persist, /vms/:id/slug
slug_available(slug) → (bool, reason?)	GET /vms/slug-available
upload_files(id, files, *, base_dir="/app")	POST /vms/:id/files
mint_widget_token(id, *, capabilities=None, ttl_seconds=None)	POST /vms/:id/widget-token
wait_until_running(id, *, timeout=300.0, interval=2.0, on_status=None)	polls GET /vms/:id

create / create_from_* accept snake_case fields: repo_url, stack, branch, subdirectory, exposed_port, install_command, start_command, environment_variables, drive_id, drive_mount_path, drive_read_only, database_integration_id, vcpus, memory_mib, persistent, slug.

client.snapshots / .drives / .integrations / .detect / .accounts / .teams / .keys

• snapshots.list() / create(vm_id, name=None) / clone(snapshot_id, *, name=None, environment_variables=None) / delete(snapshot_id)
• drives.list() / create(name, size_gib, *, mount_path=None) / delete(id)
• integrations.list() / create(**fields) / delete(id)
• databases.create(name=None, provision_api_key=None) → Database (managed Postgres; connection string revealed once)
• detect.public(repo_url, *, on_progress=None) / zip(zip_bytes, ...) / folder(path, ...)
• accounts.current() → CurrentPrincipal(account, team, auth_type, api_key)
• teams.list() / create(name, *, slug=None) / delete(team_id) → /teams
• keys.list(team_id) / create(team_id, name, *, scopes=None) / revoke(team_id, key_id) → /teams/:teamId/api-keys (create returns the full secret once)

Provision a managed database

databases.create provisions a managed Postgres and returns the connection string once — persist it immediately, it is never returned again (unlike integrations.create, which links an existing connection string and never reveals it). Requires the vms:write scope.

db = client.databases.create(name="my-app-db")
print(db.connection_string)  # revealed ONCE — store it now
print(db.id, db.host, db.database)

# Later: list databases (redacted — no connection string) or delete one.
dbs = client.integrations.list()   # kind: "database"
client.integrations.delete(db.id)

Async is identical with await: db = await client.databases.create(...).

System status

Not a dedicated resource; use the generic request helper:

from previews import SystemStatus
status = SystemStatus.from_dict(client.request("GET", "/system/status"))
print(status.caches["npm"].size_bytes)

Errors

Non-2xx responses raise PreviewsApiError(status, code, message) from the { error: { code, message } } envelope (a non-JSON body yields code UNKNOWN). wait_until_running raises PreviewsTimeoutError on deadline. Missing credentials raise PreviewsConfigError.

Models

Responses are frozen dataclasses with a tolerant from_dict (camelCase → snake_case, unknown keys ignored): Preview (alias VM), RunCommandResult, VMSnapshot, DetectionResult/EnvVarHint, BandwidthSample/BandwidthResponse, Account/Team/ApiKey/CreatedApiKey/ApiKeyRef/CurrentPrincipal, DatabaseIntegration, Database, Drive, SystemStatus/CacheStat, WidgetToken, UploadResult, LogEvent, FirewallConfig/FirewallIpRule/FirewallRateLimit. The package ships py.typed.

Widget backend proxy

The React widget UI is built separately. This SDK provides the backend piece.

Scoped token (recommended): mint a short-lived pwt_ token server-side and hand it to the browser, which then calls the platform directly.

token = client.vms.mint_widget_token(vm_id, capabilities=["preview:read", "vm:run"])

Backend proxy: the browser calls your server; the proxy forwards only a whitelisted subset of operations to the platform using the pvk_ key it holds — the key never reaches the browser.

from previews import PreviewsClient
from previews.proxy import PreviewsProxy, ProxyConfig, make_router  # or make_blueprint

client = PreviewsClient()
proxy = PreviewsProxy(ProxyConfig(
    client=client,
    allowed_origins=["https://app.example.com"],
    allowed_vm_ids={"<vm-uuid>"},          # None = any VM in the project
    # allow_ops defaults to WIDGET_SAFE_OPS = {"status","run","files","logs","mint_token"}
))

# FastAPI
from fastapi import FastAPI
app = FastAPI()
app.include_router(make_router(proxy, prefix="/previews"))

# Flask
# from flask import Flask
# app = Flask(__name__)
# app.register_blueprint(make_blueprint(proxy, url_prefix="/previews"))

The proxy enforces the origin allowlist, the op whitelist (hard-capped to the widget-safe set), and the per-VM restriction; it never echoes the pvk_ key.

License

MIT