Examples Gallery

Browse runnable examples organized by module. Each example demonstrates practical usage patterns.

Tip

All examples are standalone scripts. Run them with python examples/<module>/<script>.py

Some examples include real API credentials protected by SOPS encryption. See examples/mail/ for a working demo with auto-decrypted secrets.

---

alerts - Multi-channel alerting

Slack, Email, throttling, level filtering, and channel targeting.

Example	Description
email_basic.py	Email alerts via SMTP
slack_basic.py	Slack with channel targeting (key/alias) and batch sending
multi_channel.py	Multi-channel AlertManager with level-based filtering
kstlib.conf.yml	Config with SOPS credentials and channel aliases

cd examples/alerts

# Single channel via environment variable
export SLACK_WEBHOOK_URL="https://hooks.slack.com/services/T.../B.../xxx"
python slack_basic.py

# Multi-channel with SOPS credentials (config-driven)
python slack_basic.py --sops

Channel Targeting - target by config key or alias:

await manager.send(alert, channel="hb")         # By config key
await manager.send(alert, channel="heartbeat")   # By alias
await manager.send(alert)                        # Broadcast

Batch Sending - multiple alerts in one call:

alerts = [
    AlertMessage(title="Alert 1", body="...", level=AlertLevel.WARNING),
    AlertMessage(title="Alert 2", body="...", level=AlertLevel.WARNING),
]
results = await manager.send(alerts, channel="watchdog")

Timestamp Prefix - add send datetime to alert titles:

alert = AlertMessage(
    title="Heartbeat Check",
    body="System is healthy",
    level=AlertLevel.INFO,
    timestamp=True,  # Adds "2026-01-21 17:30:45 ::: " prefix
)

Note

The slack_basic.py example demonstrates SOPS auto-decrypt for webhook URLs. Run from examples/alerts/ directory for config auto-discovery.

auth - Authentication and OAuth2/OIDC

Keycloak and Google OAuth2 providers with SOPS token storage.

JWT Token Validation - cryptographic proof that a token is valid:

Example	Description
token_check.py	Python - kstlib TokenChecker + manual httpx/cryptography
token_check.ps1	PowerShell - raw RSA math, prints both hashes for visual proof
token_check.sh	Bash - raw RSA math via openssl, same dual-hash approach as PS1
kstlib-auth-check.sh	Bash - 6-step validation chain (curl + openssl + jq)

cd examples/auth

# Python: kstlib approach (one class, one call)
python token_check.py --verbose

# Python: manual step-by-step (httpx + cryptography)
python token_check.py --manual --verbose

# PowerShell: raw RSA, prints computed vs recovered hash
.\token_check.ps1

# Bash: raw RSA, same dual-hash proof as PowerShell
./token_check.sh
./token_check.sh --ca-bundle /etc/pki/.../tls-ca-bundle.pem  # corporate CA

# Bash: 6-step validation chain (like kstlib TokenChecker)
./kstlib-auth-check.sh --ca-bundle /path/to/ca.pem --verbose

The PowerShell and Bash token_check scripts perform raw RSA: signature^e mod n to recover the hash, then compare with SHA(header.payload). Both hashes are printed side by side for undeniable visual proof. The kstlib-auth-check.sh script mirrors the Python TokenChecker 6-step chain using only curl + openssl + jq.

Google OAuth2 - two approaches to the same flow:

Example	Description
oauth2_google.py	Production - kstlib.auth with TRACE logging (--trace)
oauth2_manual_google.py	Educational - httpx direct, see OAuth2 “under the hood”

cd examples/auth

# Production approach (~60 lines)
python oauth2_google.py your-email@gmail.com
python oauth2_google.py your-email@gmail.com --trace  # HTTP details

# Educational approach (~400 lines, explicit HTTP calls)
python oauth2_manual_google.py your-email@gmail.com

Note

Both examples use SOPS-encrypted token storage in tokens/. Tokens persist across runs and are automatically refreshed when expired. The token format is compatible between both examples.

Keycloak Providers:

File	Description
README.md	Setup instructions and usage guide
kstlib.conf.yml	Main config with provider includes

Provider configs (providers/):

Config	Description
oauth2-keycloak-dev.yml	Manual endpoints + file storage
oidc-hybrid-keycloak.yml	Hybrid mode + file storage
oidc-keycloak-dev.yml	Auto-discovery OIDC + SOPS storage

cache - Caching decorators

TTL, LRU, and file-backed caching strategies.

Example	Description
basic_usage.py	TTL cache decorator basics

config - Configuration loading

Cascading search, includes, and environment variables.

Example	Description
01_basic_usage.py	Load config from file
02_includes.py	YAML includes and merging
03_cascading_search.py	Search paths and auto-discovery
04_env_variable.py	Environment variable substitution
05_strict_format.py	Strict YAML format validation
06_error_handling.py	Error handling patterns
07_deep_merge.py	Deep merge strategies
08_multi_environment.py	Multi-environment setup
09_auto_discovery.py	Auto-discovery of config files
run_all_examples.py	Run all config examples

db - Async SQLite database

Connection pooling and SQLCipher encryption.

Example	Description
01_siren_sqlite_demo.py	In-memory SQLite for CSV analytics
02_encrypted_sqlite_demo.py	SQLCipher encrypted database

logging - Logging presets

Rich console, file rotation, and structlog integration.

Example	Description
basic_usage.py	Setup logging with presets

mail - Email builder

Templates, attachments, and multiple transport options.

Example	Description
01_config_driven.py	Config-driven transport via MailBuilder(preset=...) and mail.default
basic_plain.py	Plain text email
html_template.py	HTML with Jinja templates
attachments_inline.py	Attachments and inline images
notify_decorator.py	@mail.notify() decorator for job notifications
smtp_ethereal.py	SMTP transport with Ethereal fake service
gmail_send.py	Gmail OAuth2 transport
resend_send.py	Resend API async transport
smtp_trace.py	TRACE logging for SMTP debugging
kstlib.conf.yml	Config with SOPS include
mail.conf.sops.yml	Encrypted Resend API key

Config-driven transport - pick the transport from kstlib.conf.yml:

from kstlib.mail import MailBuilder

# Uses mail.default preset
MailBuilder().sender("bot@x.com").to("ops@x.com").message("hi").send()

# Explicit named preset
MailBuilder(preset="corporate").sender("bot@x.com").to("ops@x.com").message("hi").send()

# kstlib.conf.yml
mail:
  default: corporate
  presets:
    corporate:
      transport: smtp
      host: smtp.example.com
      port: 587
      login: user@example.com
      password: "secret"
      starttls: true
    transactional:
      transport: resend
      api_key: re_xxxxxxxxxxxxx

@mail.notify() Decorator - automatic email notifications for function execution:

from kstlib.mail import MailBuilder
from kstlib.mail.transports import SMTPTransport

transport = SMTPTransport(host="smtp.example.com", port=587)
mail = (
    MailBuilder(transport=transport)
    .sender("bot@example.com")
    .to("admin@example.com")
    .subject("ETL Job")
)

@mail.notify
def extract_data() -> dict[str, int]:
    """Extract data from source."""
    return {"rows": 1000}

# Sends "[OK] ETL Job - extract_data" on success
# Sends "[FAILED] ETL Job - extract_data" with traceback on failure
result = extract_data()

Option	Description
subject="..."	Override the base subject
on_error_only=True	Only send notification on failure
include_return=True	Include return value in email body
include_traceback=False	Exclude traceback from failure emails

Note

The notify_decorator.py example uses Ethereal Email for safe testing. Create a free account at https://ethereal.email and set ETHEREAL_USER and ETHEREAL_PASS.

Note

The resend_send.py example demonstrates SOPS auto-decrypt with real API credentials. The API key in mail.conf.sops.yml is encrypted and automatically decrypted by the config loader. Run python resend_send.py you@example.com to test (requires matching age key).

metrics - Unified metrics decorator

Timing, profiling, and call statistics.

Example	Description
01_decorators_demo.py	@metrics decorator with timing, profiling, and statistics

monitoring - HTML dashboards and reports

Typed render types, Jinja2 integration, @collector decorators, and email delivery.

Example	Description
service_dashboard.py	Render types demo with StatusCell, MonitorTable, MonitorKV
send_dashboard_gmail.py	Send dashboard via Gmail with SOPS credentials
server/run.py	Simplified API with @mon.collector decorators
server/server.monitor.yml	YAML config (template, delivery)

Simplified API - config in YAML, collectors in Python:

from kstlib.monitoring import Monitoring, MonitorKV, StatusCell, StatusLevel

mon = Monitoring.from_config()  # Load from kstlib.conf.yml

@mon.collector
def system_info():
    return MonitorKV(items={
        "Hostname": platform.node(),
        "CPU": StatusCell(f"{cpu}%", StatusLevel.OK),
    })

@mon.collector
def services():
    table = MonitorTable(headers=["Service", "Status"])
    table.add_row(["API", StatusCell("UP", StatusLevel.OK)])
    return table

mon.run_sync()  # Collect, render, deliver

Render Types - typed cells with automatic styling:

from kstlib.monitoring import (
    StatusCell, StatusLevel,
    MonitorTable, MonitorKV,
    render_template,
)

# Create status badges
api_status = StatusCell("UP", StatusLevel.OK)
db_status = StatusCell("DEGRADED", StatusLevel.WARNING)

# Create a table
table = MonitorTable(headers=["Service", "Status"])
table.add_row(["API Gateway", api_status])
table.add_row(["Database", db_status])

# Render with Jinja2
html = render_template("""
<h1>System Status</h1>
{{ table | render }}
""", {"table": table})

Email Delivery - send dashboards via Gmail OAuth2:

# kstlib.conf.yml
monitoring:
  template_file: dashboard.html
  inline_css: true
  delivery:
    type: mail
    sender: bot@example.com
    recipients: [team@example.com]

cd examples/monitoring/server
python run.py              # Send email
python run.py --no-send    # Save to file only

Note

Mail delivery requires Gmail OAuth2 token. Run kstlib auth login google first.

pipeline - Declarative pipeline execution

Shell commands, Python modules, and callable functions in sequential workflows.

Example	Description
01_basic_shell.py	Simple shell steps (echo, date, platform)
02_multi_step.py	Conditional steps with on_success/on_failure
03_config_driven.py	Pipeline loaded from kstlib.conf.yml
kstlib.conf.yml	Example pipeline config

cd examples/pipeline
python 01_basic_shell.py     # Basic shell pipeline
python 02_multi_step.py      # Conditions + callable
python 03_config_driven.py   # Config-driven (from_config)

Step Types - three execution modes:

from kstlib.pipeline import StepConfig, StepType

# Shell: subprocess.run(shell=True)
StepConfig(name="build", type=StepType.SHELL, command="make build")

# Python: python -m module
StepConfig(name="lint", type=StepType.PYTHON, module="ruff", args=("check",))

# Callable: importlib import + call
StepConfig(name="notify", type=StepType.CALLABLE, callable="my.alerts:send_ok")

Conditional Steps - skip or run based on previous results:

from kstlib.pipeline import StepCondition

# Only runs if all previous steps succeeded
StepConfig(name="deploy", ..., when=StepCondition.ON_SUCCESS)

# Only runs if a previous step failed (cleanup)
StepConfig(name="cleanup", ..., when=StepCondition.ON_FAILURE)

ops - Unified session management

tmux and containers (Docker/Podman), config-driven via kstlib.conf.yml.

Config-Driven Examples:

Folder	Description
config_driven/tmux/	tmux session from config
config_driven/container/	Docker/Podman container from config
config_driven/README.md	Lifecycle, auto-detect, CLI overrides

Resilience Test (Long-Running):

File	Description
resilience_tmux/README.md	Architecture and usage guide
resilience_tmux/main.py	Binance 1h candles + 4h proactive restart
resilience_tmux/candle_logger.py	CSV logger with gap analysis
resilience_tmux/kstlib.conf.yml	Config with resilience + Slack alerts

Components demonstrated: SessionManager, BinanceKlineStream, TimeTrigger("240m"), Heartbeat + StateWriter, CandleLogger, AlertManager with Slack, GracefulShutdown.

cd examples/ops/resilience_tmux

# Direct run (foreground)
python main.py

# Via kstlib ops (tmux session, detachable)
kstlib ops start resilience-tmux
kstlib ops attach resilience-tmux   # Ctrl+B D to detach
kstlib ops logs resilience-tmux
kstlib ops stop resilience-tmux

CLI Quick Reference:

kstlib ops start <name>              # Start from config
kstlib ops stop <name>               # Graceful stop
kstlib ops attach <name>             # Interactive attach
kstlib ops status <name> [--json]    # Session state
kstlib ops logs <name> [--lines 50]  # Recent output
kstlib ops list [--backend tmux]     # All sessions

rapi - Config-driven REST API client

Define APIs in YAML, call from CLI or Python.

Basic Examples:

Example	Description
01_httpbin_basics.py	Basic usage with httpbin.org (GET, POST, path params, headers)
02_credentials_and_auth.py	Credential resolver, Bearer/Basic auth
03_error_handling.py	HTTP errors, timeouts, retry behavior

External Config Examples - real-world with *.rapi.yml config files:

Folder	Description
binance/	Binance API - HMAC signing, SOPS credentials, testnet
github/	GitHub API - Real token via SOPS, from_file() demo
httpbin/	httpbin.org - External config, auto-discovery
viya/	SAS Viya - POC with ~1250 endpoints, nested includes

# Run the Binance demo (requires SOPS-encrypted API keys)
cd examples/rapi/binance
kstlib rapi binance.balance       # Account info (HMAC signed)
kstlib rapi binance.ticker-price  # Public endpoint (no auth)

# Run the GitHub demo (requires SOPS-encrypted token)
cd examples/rapi/github
python demo.py

CLI Examples:

kstlib rapi list                                    # List endpoints
kstlib rapi github.user                             # Simple GET
kstlib rapi github.repos-issues owner=KaminoU repo=igcv3  # Path params
kstlib rapi github.repos-list per_page=50 page=2    # Query params
kstlib rapi myapi.create -b '{"user": "alice"}'     # POST with JSON
kstlib rapi myapi.create -b @data.json              # Body from file
kstlib rapi github.user -o user.json                # Output to file
kstlib -vvv rapi github.user                        # TRACE mode

resilience - Fault tolerance patterns

Circuit breaker, graceful shutdown, heartbeat, rate limiter, and watchdog.

Example	Description
01_circuit_breaker.py	Circuit breaker pattern with state machine
02_graceful_shutdown.py	Priority-based shutdown callbacks
03_heartbeat.py	Process liveness signaling
04_rate_limiter.py	Token bucket rate limiting
05_watchdog.py	Thread/process freeze detection

binance_testnet/ - complete real-world integration for a Binance WebSocket bot:

File	Description
README.md	Full documentation with architecture diagram
main.py	Main application with dashboard
watchdog_service.py	External watchdog process
kstlib.conf.yml	Configuration with includes

Components demonstrated: Heartbeat with HeartbeatTarget, Watchdog.from_state_file(), TimeTrigger, AlertManager with SlackChannel, WebSocketManager with managed reconnection.

cd examples/resilience/binance_testnet
python main.py

secrets - Secret resolution

SOPS encryption with AGE and AWS KMS backends.

AGE Examples (local encryption):

Example	Description
decrypt_example.py	Decrypt secrets with AGE
decrypt_secure_example.py	Secure decryption patterns
encrypt_example.py	Encrypt secrets with SOPS

AWS KMS Examples (production-grade encryption):

Example	Description
kms/.sops.yaml	SOPS config with real AWS KMS key
kms/secrets.example.yml	Template for KMS-encrypted secrets
kms/decrypt_example.py	Decrypt KMS-encrypted secrets

# Decrypt via CLI (auto-detects KMS from sops metadata)
kstlib secrets decrypt examples/secrets/kms/secrets.sops.yml

# Or via Python
from kstlib.config.sops import decrypt_sops_file
secrets = decrypt_sops_file("secrets.sops.yml")

Note

The kms/ examples use a real production AWS KMS key in eu-west-3. This demonstrates enterprise-grade secret management with automatic key rotation and IAM-based access control. Requires valid AWS credentials with KMS permissions.

secure - Filesystem guardrails

Path validation and secure file operations.

Example	Description
guardrails_demo.py	Path validation and permission checks

transform - Bidirectional data transformation

Encode/decode/patch round-trips for nested binary blobs (base64, zlib, JSON, XML). Composed patches with global + targeted filters.

Example	Description
01_round_trip.py	Programmatic chain build, full forward + patch + backward, integrity check
02_config_driven.py	Load chain from kstlib.conf.yml via TransformChain.from_config()
03_composed_patch.py	global + targeted patches with filter cascade, plus scope: all outer mutation for R220
04_outer_patch.py	The three scope: values (blob, outer, all), replace_outer_uris helper, custom protected paths
kstlib.conf.yml	Preset + usage chains, replace + callable patches, composed cascade with scope: blob / scope: all

cd examples/transform
python 01_round_trip.py     # Programmatic round-trip with synthetic data
python 02_config_driven.py  # Config-driven via from_config()
python 03_composed_patch.py # Composed patch cascade on 3 synthetic objects
python 04_outer_patch.py    # scope: blob | outer | all on a synthetic wrapper

Round-trip pattern - decode, patch, re-encode with lossless envelope:

from kstlib.transform import (
    TransformChain,
    TransformChainConfig,
    PrimitiveConfig,
    PatchConfig,
)

chain = TransformChain(
    TransformChainConfig(
        name="decode_report",
        forward=(
            PrimitiveConfig(name="base64"),
            PrimitiveConfig(name="zlib"),
            PrimitiveConfig(
                name="json",
                options={"extract": "transferableContent.content"},
            ),
        ),
        patch=PatchConfig(replace={"old-host": "new-host"}),
    )
)

# One-shot: forward + patch + backward
patched_blob = chain.transform(blob_b64_string)

Composed patches - apply different patches to different objects:

# kstlib.conf.yml
transforms:
  chains:
    patch_production:
      preset: sas_report          # Inherit forward + backward

      global_patches:
        - patch_report            # Applied to every object (scope: blob)

      targeted_patches:
        - filter:
            content_type: report
            name: "R220_*"
          patches:
            - patch_report_full   # R220 reports also patch the outer wrapper
        - filter:
            name: "*"             # Wildcard fallback
          patches:
            - patch_report

import json
from kstlib.transform import transform

# Pass metadata so the targeted_patches filters can match. R220 reports
# also need metadata['outer'] because patch_report_full has scope: all.
wrapper = json.loads(transfer_object_json)
result = transform(
    wrapper["content"],
    "patch_production",
    metadata={
        "content_type": "report",
        "name": "R220_ASTRO",
        "outer": wrapper,
    },
)

Note

For the full user guide (5 primitives, auto-reverse rules, security hardening, YAML schema reference), see Transform. For the API reference (autoclass + autofunction), see Bidirectional Data Transformation.

ui - Rich-based UI components

Panels, spinners, and tables for terminal output.

Panels:

Example	Description
panel_basic_usage.py	Panel presets and customization

Spinners - animated feedback for CLI operations:

Example	Description
01_basic_usage.py	Context manager and manual control
02_all_styles.py	Gallery of 9 spinner styles
03_animation_types.py	Spin, bounce, and color wave
04_presets.py	Built-in presets and overrides
05_customization.py	Styling, position, speed
06_with_logs.py	Logs scrolling above spinner
07_decorator_and_zones.py	@with_spinner decorator + log zones
run_all_examples.py	Run all spinner demos

Tables:

Example	Description
table_basic_usage.py	Table builder with column config

---

Running Examples

Single example

python examples/resilience/01_circuit_breaker.py
python examples/ui/spinner/01_basic_usage.py

All examples in a module

python examples/config/run_all_examples.py
python examples/ui/spinner/run_all_examples.py

From project root with module syntax

python -m examples.ui.spinner.01_basic_usage