Skip to main content

Python SDK for building stateful processors with zero-downtime migration, auto-initialization, and smart instrumentation

Project description

Dory SDK for Python

A Python SDK for building stateful processors with zero-downtime migration, graceful shutdown, and state transfer on Kubernetes.

What Does This SDK Do For You?

Feature Without Dory SDK With Dory SDK
Pod shutdown App killed, state lost State saved automatically, restored on new pod
Node maintenance Downtime, manual intervention Zero-downtime migration to new node
Crash recovery Start from scratch Resume from last checkpoint
Health monitoring DIY implementation Built-in /healthz, /ready, /metrics

Quick Start (Choose Your Style)

Option A: Minimal (7 lines)

from dory import DoryApp, BaseProcessor, stateful

class MyApp(BaseProcessor):
    counter = stateful(0)  # Auto-saved and restored!

    async def run(self):
        async for _ in self.run_loop(interval=1):
            self.counter += 1

if __name__ == "__main__":
    DoryApp().run(MyApp)

Option B: Function-Based (6 lines)

from dory.simple import processor, state

counter = state(0)

@processor
async def main(ctx):
    async for _ in ctx.run_loop(interval=1):
        counter.value += 1

Option C: Full Control

from dory import DoryApp, BaseProcessor, ExecutionContext

class MyApp(BaseProcessor):
    def __init__(self, context: ExecutionContext):
        super().__init__(context)
        self.counter = 0

    async def startup(self):
        self.context.logger().info("Starting...")

    async def run(self):
        while not self.context.is_shutdown_requested():
            self.counter += 1
            await asyncio.sleep(1)

    async def shutdown(self):
        self.context.logger().info(f"Final count: {self.counter}")

    def get_state(self):
        return {"counter": self.counter}

    async def restore_state(self, state):
        self.counter = state.get("counter", 0)

if __name__ == "__main__":
    DoryApp().run(MyApp)

Installation

pip install dory-sdk[kubernetes]

CLI Tool

The SDK includes a CLI for generating Kubernetes manifests:

# Initialize a new project with all files
dory init my-app --image my-app:latest

# Output:
#   Created main.py
#   Created Dockerfile
#   Created k8s/rbac.yaml
#   Created k8s/deployment.yaml

# Generate specific manifests
dory generate rbac --name my-app
dory generate deployment --name my-app --image my-app:latest
dory generate all --name my-app --image my-app:latest

# Validate configuration
dory validate

Deployment Options

Option 1: Helm Chart

helm install my-app ./helm/dory-processor \
  --set name=my-app \
  --set image.repository=my-app \
  --set image.tag=latest

With values file:

helm install my-app ./helm/dory-processor -f values.yaml

Option 2: Kustomize

# Deploy to dev
kubectl apply -k kustomize/overlays/dev

# Deploy to production
kubectl apply -k kustomize/overlays/production

Option 3: CLI Generated Manifests

dory init my-app --image my-app:latest
kubectl apply -f k8s/

When to Use Which

Choose When
Helm Need release management, rollback, existing Helm workflow
Kustomize GitOps (ArgoCD/Flux), prefer patch-based config
CLI Quick start, simple deployments

Sidecar Mode (No SDK Required)

Don't want to integrate the SDK? Use sidecar mode - your app runs unchanged, a lightweight sidecar handles Kubernetes health endpoints.

What You Get

Feature With SDK Sidecar Mode
Health endpoints Yes Yes
Graceful shutdown Yes Yes
State migration Yes No
Zero code changes No Yes

Deploy with Sidecar

Using Helm:

helm install my-app ./helm/dory-processor \
  --set image.repository=your-app \
  --set image.tag=latest \
  --set sidecar.enabled=true

Using Kustomize:

kubectl apply -k kustomize/overlays/sidecar

How It Works

┌─────────────────────────────────────────┐
│                  Pod                     │
│  ┌─────────────┐    ┌────────────────┐  │
│  │  Your App   │    │  Dory Sidecar  │  │
│  │  (no SDK)   │    │                │  │
│  │             │    │  /healthz ←────┼──┼── K8s liveness
│  │  port 8081 ←┼────┼→ /ready   ←────┼──┼── K8s readiness
│  │             │    │  /prestop ←────┼──┼── K8s preStop
│  │             │    │  /metrics      │  │
│  └─────────────┘    └────────────────┘  │
└─────────────────────────────────────────┘

The sidecar:

  • Responds to Kubernetes health probes
  • Optionally monitors your app's port/health endpoint
  • Handles graceful shutdown signals

Sidecar Configuration

Environment Variable Default Description
DORY_APP_PORT - Your app's port (optional monitoring)
DORY_APP_HEALTH_PATH - Your app's health endpoint to check
DORY_APP_PRESTOP_PATH - Your app's shutdown endpoint to call
DORY_READY_REQUIRES_APP false Fail /ready if app doesn't respond

Build the Sidecar Image

docker build -f Dockerfile.sidecar -t dory-sidecar:1.0.0 .

Integration Guide

Step 1: Install SDK

pip install dory-sdk[kubernetes]

Step 2: Write Your Processor

Minimal (with @stateful):

from dory import DoryApp, BaseProcessor, stateful

class MyApp(BaseProcessor):
    # These are automatically saved/restored
    counter = stateful(0)
    data = stateful(dict)

    async def run(self):
        async for i in self.run_loop(interval=1):
            self.counter += 1

if __name__ == "__main__":
    DoryApp().run(MyApp)

That's it! The SDK handles:

  • get_state() - auto-generated from @stateful vars
  • restore_state() - auto-generated from @stateful vars
  • startup() - default no-op
  • shutdown() - default no-op

Step 3: Deploy to Kubernetes

Option A: Use CLI

dory init my-app --image my-app:latest
kubectl apply -f k8s/

Option B: Use Helm

helm install my-app ./helm/dory-processor --set image.repository=my-app

Option C: Use Kustomize

# Edit kustomize/overlays/dev/kustomization.yaml to set your image
kubectl apply -k kustomize/overlays/dev

Option D: Manual Setup

  1. Create RBAC:
apiVersion: v1
kind: ServiceAccount
metadata:
  name: my-app
---
apiVersion: rbac.authorization.k8s.io/v1
kind: Role
metadata:
  name: my-app-state-manager
rules:
- apiGroups: [""]
  resources: ["configmaps"]
  verbs: ["get", "create", "update", "patch", "delete"]
---
apiVersion: rbac.authorization.k8s.io/v1
kind: RoleBinding
metadata:
  name: my-app-state-manager
subjects:
- kind: ServiceAccount
  name: my-app
roleRef:
  kind: Role
  name: my-app-state-manager
  apiGroup: rbac.authorization.k8s.io
  1. Create Deployment:
apiVersion: apps/v1
kind: Deployment
metadata:
  name: my-app
spec:
  replicas: 1
  selector:
    matchLabels:
      app: my-app
  template:
    metadata:
      labels:
        app: my-app
    spec:
      serviceAccountName: my-app
      terminationGracePeriodSeconds: 35
      containers:
      - name: my-app
        image: my-app:latest
        env:
        - name: DORY_POD_NAME
          valueFrom:
            fieldRef:
              fieldPath: metadata.name
        - name: DORY_POD_NAMESPACE
          valueFrom:
            fieldRef:
              fieldPath: metadata.namespace
        livenessProbe:
          httpGet:
            path: /healthz
            port: 8080
        readinessProbe:
          httpGet:
            path: /ready
            port: 8080
        lifecycle:
          preStop:
            httpGet:
              path: /prestop
              port: 8080

Features

@stateful Decorator

Mark variables for automatic state management:

from dory import BaseProcessor, stateful

class MyApp(BaseProcessor):
    # Simple values
    counter = stateful(0)
    name = stateful("default")

    # Mutable defaults (use factory)
    data = stateful(dict)    # Creates new dict for each instance
    items = stateful(list)   # Creates new list for each instance

    async def run(self):
        # Just use them normally - SDK handles save/restore
        self.counter += 1
        self.data["key"] = "value"

run_loop() Helper

Simplifies the shutdown check pattern:

# Instead of:
async def run(self):
    while not self.context.is_shutdown_requested():
        self.counter += 1
        await asyncio.sleep(1)

# Use:
async def run(self):
    async for i in self.run_loop(interval=1):
        self.counter += 1
        print(f"Iteration {i}")

Function-Based API

For simple apps that don't need a class:

from dory.simple import processor, state

counter = state(0)
sessions = state(dict)

@processor
async def main(ctx):
    logger = ctx.logger()

    async for i in ctx.run_loop(interval=1):
        counter.value += 1
        logger.info(f"Count: {counter.value}")

ExecutionContext

Access pod metadata and utilities:

async def run(self):
    ctx = self.context

    # Logging with pod context
    ctx.logger().info("Processing...")

    # Pod metadata
    print(f"Pod: {ctx.pod_name}")
    print(f"Namespace: {ctx.pod_namespace}")
    print(f"Processor ID: {ctx.processor_id}")
    print(f"Restart count: {ctx.attempt_number}")

    # App config (env vars except DORY_*)
    config = ctx.config()
    model_path = config.get("MODEL_PATH")

    # Shutdown detection
    while not ctx.is_shutdown_requested():
        if ctx.is_migration_imminent():
            print("Migration coming, finishing batch...")
        await process()

Configuration

Environment Variables

Variable Default Description
DORY_HEALTH_PORT 8080 Health server port
DORY_LOG_LEVEL INFO Log level
DORY_LOG_FORMAT json Log format (json/text)
DORY_STATE_BACKEND configmap State storage backend
DORY_STARTUP_TIMEOUT_SEC 30 Startup timeout
DORY_SHUTDOWN_TIMEOUT_SEC 30 Shutdown timeout

Config File (dory.yaml)

health_port: 8080
log_level: INFO
log_format: json
state_backend: configmap
startup_timeout_sec: 30
shutdown_timeout_sec: 30

Local Development

Test locally without Kubernetes:

DORY_STATE_BACKEND=local python main.py

HTTP Endpoints

Endpoint Description
GET /healthz Liveness probe (200=alive)
GET /ready Readiness probe (200=ready)
GET /metrics Prometheus metrics
GET /state Get processor state
POST /state Restore processor state
GET /prestop PreStop hook handler

API Reference

BaseProcessor Methods

Method Required Description
run() Yes Main processing loop
startup() No Initialize resources (default: no-op)
shutdown() No Cleanup resources (default: no-op)
get_state() No Return state dict (default: @stateful vars)
restore_state(state) No Restore state (default: @stateful vars)

Helper Methods

Method Description
run_loop(interval) Async iterator with auto shutdown check
is_shutting_down() Check if shutdown requested

Fault Handling Hooks (Optional)

Method Description
on_state_restore_failed(error) Handle restore errors
on_rapid_restart_detected(count) Handle restart loops
on_health_check_failed(error) Handle health failures
reset_caches() Clear caches on golden reset

How State Migration Works

Pod Shutdown

1. Kubernetes calls /prestop
2. SDK saves state to ConfigMap
3. Pod marked not-ready
4. Your run() exits
5. Your shutdown() called
6. Pod terminates

New Pod Startup

1. SDK finds state in ConfigMap
2. Your startup() called
3. Your restore_state() called
4. Pod marked ready
5. Your run() starts

Comparison: Before vs After

Before (25+ lines)

import asyncio
from dory import DoryApp, BaseProcessor, ExecutionContext

class MyApp(BaseProcessor):
    def __init__(self, context: ExecutionContext):
        super().__init__(context)
        self.counter = 0
        self.sessions = {}

    async def startup(self) -> None:
        pass

    async def run(self) -> None:
        while not self.context.is_shutdown_requested():
            self.counter += 1
            await asyncio.sleep(1)

    async def shutdown(self) -> None:
        pass

    def get_state(self) -> dict:
        return {"counter": self.counter, "sessions": self.sessions}

    async def restore_state(self, state: dict) -> None:
        self.counter = state.get("counter", 0)
        self.sessions = state.get("sessions", {})

if __name__ == "__main__":
    DoryApp().run(MyApp)

After (7 lines)

from dory import DoryApp, BaseProcessor, stateful

class MyApp(BaseProcessor):
    counter = stateful(0)
    sessions = stateful(dict)

    async def run(self):
        async for _ in self.run_loop(interval=1):
            self.counter += 1

if __name__ == "__main__":
    DoryApp().run(MyApp)

Documentation

Examples

Example Description Pattern
minimal-processor-py Simplest possible processor (~95 lines) @stateful + run_loop()
dory-info-logger-py Full demo with HTTP dashboard @stateful + fault hooks
dory-edge-logger-py Edge workload with DB logging Manual state management

Start here: Use minimal-processor-py as a template for new processors.

License

Apache 2.0

Project details


Download files

Download the file for your platform. If you're not sure which to choose, learn more about installing packages.

Source Distribution

dory_sdk-2.1.0.tar.gz (126.1 kB view details)

Uploaded Source

Built Distribution

If you're not sure about the file name format, learn more about wheel file names.

dory_sdk-2.1.0-py3-none-any.whl (126.6 kB view details)

Uploaded Python 3

File details

Details for the file dory_sdk-2.1.0.tar.gz.

File metadata

  • Download URL: dory_sdk-2.1.0.tar.gz
  • Upload date:
  • Size: 126.1 kB
  • Tags: Source
  • Uploaded using Trusted Publishing? No
  • Uploaded via: twine/6.2.0 CPython/3.10.0

File hashes

Hashes for dory_sdk-2.1.0.tar.gz
Algorithm Hash digest
SHA256 bf9c8dbf381cc49d3461ec8c54e5c8e51ddc7d4aed496fe07a1769794b0245ec
MD5 f5386c28d0ef7d1990b338e97b5e31d1
BLAKE2b-256 dbe19cbdd18f46636e95d749908379fb2c0f710c269900444cfa07beb1a20261

See more details on using hashes here.

File details

Details for the file dory_sdk-2.1.0-py3-none-any.whl.

File metadata

  • Download URL: dory_sdk-2.1.0-py3-none-any.whl
  • Upload date:
  • Size: 126.6 kB
  • Tags: Python 3
  • Uploaded using Trusted Publishing? No
  • Uploaded via: twine/6.2.0 CPython/3.10.0

File hashes

Hashes for dory_sdk-2.1.0-py3-none-any.whl
Algorithm Hash digest
SHA256 cf35affb9b637d3440dd307bdd6f280949c9227af04ff4bc1e8d0a7e1815b1b5
MD5 efda4965ad097a8657c9bbb4bbc1d958
BLAKE2b-256 1a718e8efae17d58cf857f51c282980cb90cae011028097df2b16a76731ce995

See more details on using hashes here.

Supported by

AWS Cloud computing and Security Sponsor Datadog Monitoring Depot Continuous Integration Fastly CDN Google Download Analytics Pingdom Monitoring Sentry Error logging StatusPage Status page