SOASAP Python SDK

Overview

The SOASAP Python SDK synchronizes feature flag configuration from SOASAP Cloud in the background and evaluates flags from an in-memory snapshot. Flag reads are synchronous, O(1), and never perform network or disk I/O on the hot path.

The SDK provides:

• Local evaluation — dictionary lookups against an immutable snapshot
• Real-time synchronization — Server-Sent Events (SSE) delta updates
• Persistent cache — disk-backed snapshot for cold starts
• Offline operation — last known snapshot when the network or control plane is unavailable
• Explicit defaults — every getter accepts a code-defined fallback
• Non-blocking startup — optional preload=True starts sync without blocking import or server boot

Runtime characteristics

• O(1) local evaluation — reads from an in-memory snapshot; no network requests during evaluation
• Real-time SSE synchronization — background worker applies validated payloads with atomic configuration updates
• Persistent cache — disk-backed snapshot for cold starts and restarts
• Offline operation — last known snapshot when the network or control plane is unavailable
• Explicit defaults — missing, mistyped, or unsynced keys return your code-defined fallback
• Non-blocking startup — preload=True starts synchronization without blocking module initialization

Requirements

Supported runtimes:

• Python 3.10, 3.11, 3.12, 3.13+

Supported application types:

• Flask, FastAPI, Django, and other WSGI/ASGI frameworks
• Celery workers and background job processors
• CLI tools and standalone scripts
• Gunicorn, uWSGI, and uvicorn deployments (one client per worker process)

Runs on Linux, Windows, and macOS. Requires outbound HTTPS to api.soasap.com for SSE synchronization.

Installation

pip:

pip install soasap

uv:

uv add soasap

Poetry:

poetry add soasap

pyproject.toml dependency:

[project]
dependencies = [
  "soasap==1.0.3",
]

Verify installation

pip show soasap

Confirm soasap appears in the installed package list before continuing.

Before you begin

You will need:

• ✓ A SOASAP account
• ✓ A project
• ✓ An environment
• ✓ An environment API key

If you have not completed these steps, start with Quick Start.

Installation flow

After your project and environment exist in the dashboard, install the PyPI package and call create_soasap_client once at application startup. The API key binds the process to a single environment. Synchronization runs in a background thread; request handlers evaluate flags locally.

Client initialization

Create one client per process and enable preload for production servers:

import os
from soasap import create_soasap_client

api_key = os.environ.get("SOASAP_API_KEY")
if not api_key:
    raise RuntimeError("SOASAP_API_KEY is required.")

flags = create_soasap_client(
    api_key=api_key,
    preload=True,
)

• API key — environment-scoped secret sent as X-API-Key on SSE requests; identifies which configuration to synchronize
• One client per process — reuse the same SoasapClient instance across your application (one per Gunicorn worker)
• Preload — with preload=True, the SSE worker starts in a daemon thread during construction without blocking import

Attach the client to your request context (Flask.g, FastAPI app.state, or a module-level singleton) so handlers can read flags synchronously.

Configuration

Pass options to create_soasap_client. The SDK does not load .env files automatically — use python-dotenv or your framework's settings module if needed.

Environment variable:

SOASAP_API_KEY=soasap...

flags = create_soasap_client(
    api_key=os.environ["SOASAP_API_KEY"],
    preload=True,
)

Using SoasapClientOptions:

from soasap import SoasapClientOptions, create_soasap_client

flags = create_soasap_client(
    SoasapClientOptions(
        api_key=os.environ["SOASAP_API_KEY"],
        preload=True,
    )
)

Optional configuration:

from soasap import SoasapErrorSource, create_soasap_client

flags = create_soasap_client(
    api_key=os.environ["SOASAP_API_KEY"],
    base_url="https://api.soasap.com",
    cache_directory="/var/lib/myapp/soasap-cache",
    preload=True,
    on_error=lambda ctx: print(
        f"[SOASAP {ctx.source.value}] transient={ctx.is_transient} {ctx.exception}"
    ),
)

Evaluating flags

All getters read from the current in-memory snapshot. Wrong JSON types return the default you provide.

Boolean

enabled = flags.get_bool("new-checkout", False)

get_flag is an alias for get_bool.

String

theme = flags.get_string("checkout-theme", "default")

Number

SOASAP number flags map to Python float. Use for integers, limits, rates, and thresholds.

max_items = flags.get_number("max-items", 10)
discount_rate = flags.get_number("discount-rate", 0.0)

JSON

from typing import TypedDict

class CheckoutSettings(TypedDict):
    enabled: bool
    timeout_seconds: int

settings = flags.get_json(
    "checkout-config",
    {"enabled": False, "timeout_seconds": 30},
)

JSON keys from the dashboard are returned as stored (typically snake_case or camelCase depending on how flags are configured). get_json returns a deep copy of dict or list values. Non-object values or copy failures return the default; getters do not throw.

Explicit defaults

Always pass a default on every read:

flags.get_bool("new-checkout", False)

The default is returned when:

• The flag key does not exist in the synchronized snapshot
• The flag was deleted or disabled in the dashboard for this environment
• The process starts before the first SSE payload arrives and no disk cache exists
• A restored disk cache is empty or invalid
• The flag value type does not match the getter (for example, a string flag read with get_bool)
• Synchronization is temporarily delayed after deploy or reconnect

Explicit defaults are your production safety net. They define safe behavior when configuration is missing, stale, or not yet synchronized — without try/except on the hot path.

Runtime architecture

• SSE delivers full or delta flag payloads to a background daemon thread
• Valid payloads replace the in-memory snapshot atomically — readers never observe partial updates
• Evaluation reads a local reference to an immutable snapshot with no locks on the request path
• No HTTP requests and no blocking disk I/O during get_bool, get_number, get_string, or get_json

Implementation details on the hot path:

• Snapshot replacement swaps an object reference — concurrent readers keep the previous snapshot until the swap completes
• Each getter copies the current snapshot reference locally before lookup
• SSE parser enforces a 5 MB payload cap; invalid root payloads are ignored

Persistent cache

On construction, the client synchronously loads the last persisted JSON snapshot from disk before any network activity. SSE updates enqueue debounced disk writes (coalesced every ~2.5 seconds).

Default cache locations:

Override with cache_directory="/custom/path". Each API key maps to a separate cache file (hash derived from the key).

Deploy and restart — rolling deploys and worker restarts hydrate from disk immediately; SSE reconnects in the background. First deploy on a new machine without cache returns defaults until the first successful sync.

Offline operation

The SDK continues evaluating flags from the last known snapshot when:

• Internet connectivity is unavailable
• SOASAP Cloud returns errors or is unreachable
• The SSE connection drops (proxy timeout, firewall, transient network failure)

The background worker reconnects with exponential backoff. Reads are unaffected during disconnect — only future updates are delayed.

Stale configuration is preferable to unavailable configuration: the application keeps running with the last known values and your code defaults.

Preload and startup behavior

preload=True is recommended for production WSGI/ASGI servers and long-running workers.

With preload:

1. Client construction loads the disk cache synchronously
2. The SSE worker starts in a daemon thread — no blocking during import or app factory execution
3. The server keeps booting while synchronization completes in the background
4. First requests may read disk cache or explicit defaults until the first SSE payload arrives

Without preload (lazy mode):

flags = create_soasap_client(api_key=os.environ["SOASAP_API_KEY"])

The SSE worker starts on the first flag read. Until then, evaluation uses the disk cache (if present) or your explicit defaults. Prefer preload for predictable synchronization timing after deploy.

Graceful shutdown

Call close() when the process exits to stop the SSE thread, flush the disk cache (up to 5 seconds), and release resources:

import atexit

flags = create_soasap_client(
    api_key=os.environ["SOASAP_API_KEY"],
    preload=True,
)
atexit.register(flags.close)

In FastAPI, prefer a lifespan handler. In Gunicorn, register shutdown hooks per worker process.

Best practices

Common issues

Flag always returns the default

• Verify the API key matches the intended environment
• Confirm the flag key exists and is enabled in that environment
• Check the flag type matches the getter (get_bool for boolean flags)

Dashboard changes do not appear

• Verify outbound HTTPS to api.soasap.com and SSE is not blocked by a proxy or firewall
• See SSE Disconnected

Wrong value returned

• Verify the flag value in the selected environment — Development and Production are independent
• Confirm the deployed process uses the API key for that environment

Multiple SDK instances

Symptoms

• Multiple SSE connections from the same worker process
• Increased memory usage
• Duplicate synchronization traffic

Resolution — call create_soasap_client once per worker and reuse the same client throughout the application. Do not instantiate a new client inside view functions or per-request middleware.

Startup synchronization delays

Expected with preload: the process starts immediately; the first SSE payload may arrive seconds after boot. Until then, evaluation uses the disk cache or explicit defaults. This is by design — startup is never blocked on sync.

Complete example

The following example demonstrates the recommended production setup for FastAPI applications, including configuration, client initialization, preload, lifespan shutdown, and runtime flag evaluation.

import os
from contextlib import asynccontextmanager

from fastapi import FastAPI
from soasap import create_soasap_client

api_key = os.environ.get("SOASAP_API_KEY")
if not api_key:
    raise RuntimeError("Set SOASAP_API_KEY via environment variable.")

flags = create_soasap_client(api_key=api_key, preload=True)


@asynccontextmanager
async def lifespan(app: FastAPI):
    yield
    flags.close()


app = FastAPI(lifespan=lifespan)


@app.get("/checkout")
def checkout():
    if not flags.get_bool("new-checkout", False):
        return {"mode": "legacy"}

    return {
        "mode": "new",
        "max_items": int(flags.get_number("max-items", 10)),
        "theme": flags.get_string("checkout-theme", "default"),
    }


@app.get("/health")
def health():
    if flags.get_bool("maintenance-mode", False):
        return "Maintenance", 503
    return "OK"

Production recommendations

Next steps

• Feature Flags — flag types, defaults, and naming
• Architecture — local evaluation, SSE, cache, and offline operation
• API Keys — issuing, rotating, and securing environment keys
• Other SDKs — .NET, Node.js, React, and more

Deep dives: Initialization, Cache, Offline Behavior.