dory-processor-sdk 0.0.2

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

Dory Processor SDK

A production-ready Python SDK for building stateful, fault-tolerant processors on Kubernetes with zero-downtime migration, automatic state persistence, and comprehensive observability.

Part of the Dory platform. The Dory Processor SDK is the official Python SDK for building processors that run on the Dory platform. It is designed to work together with the Dory orchestrator, which manages pod lifecycle, zero-downtime migration, state transfer, and edge failover for your processors.

Why Dory Processor SDK?

Challenge	Without Dory	With Dory Processor SDK
Pod termination	State lost, start from scratch	State auto-saved to ConfigMap, restored on new pod
Node drain/maintenance	Downtime, manual intervention	Zero-downtime migration with state transfer
Transient failures	DIY retry logic	Built-in retry with exponential backoff
Cascading failures	Service degradation	Circuit breakers protect dependencies
Debugging distributed systems	Scattered logs	OpenTelemetry tracing across services
Health monitoring	Custom implementation	Built-in /health, /ready, /metrics

Features

Core

• @stateful decorator - Automatic state persistence and restoration
• DoryApp - Application lifecycle management
• BaseProcessor - Base class with built-in hooks
• ExecutionContext - Pod metadata, logging, shutdown detection

Resilience

• Circuit Breaker - Prevent cascading failures with configurable thresholds
• Retry with Backoff - Exponential backoff with jitter and retry budgets
• Error Classification - Intelligent error categorization (transient, permanent, resource)

Observability

• OpenTelemetry - Distributed tracing with automatic span creation
• Prometheus Metrics - Built-in /metrics endpoint
• Structured Logging - JSON logs with request context

State Management

• ConfigMap Backend - Persist state in Kubernetes ConfigMaps
• S3 Backend - Store large state in S3 (with offline buffering)
• Local Backend - File-based storage for development
• State Versioning - Forward/backward compatible state formats

Recovery

• Restart Detection - Detect rapid restart loops
• State Validation - Validate state integrity before restore
• Golden Snapshots - Create known-good state checkpoints
• Partial Recovery - Recover individual fields when full restore fails
• Golden Image Reset - Factory reset capability

Middleware

• Request Tracking - Track request lifecycle and metrics
• Request ID Propagation - Automatic request ID across service calls
• Connection Tracking - Monitor connection lifecycle

Edge Support

• Fencing Manager - Split-brain prevention for edge deployments
• Heartbeat Manager - Connectivity monitoring
• Role Manager - Primary/standby failover

Installation

pip install dory-processor-sdk                 # Core SDK (aiohttp, pydantic, PyYAML)
pip install dory-processor-sdk[production]     # EKS deployment (K8s, S3, OpenTelemetry, RabbitMQ, Redis)
pip install dory-processor-sdk[dev]            # Test/lint tooling (pytest, mypy, ruff, black)

Only two optional-dependency extras exist: [production] and [dev]. The Kubernetes, S3 (boto3), OpenTelemetry, RabbitMQ (aio-pika) and Redis dependencies are all bundled inside [production].

Quick Start

Minimal Example (7 lines)

from dory import DoryApp, BaseProcessor, stateful

class MyApp(BaseProcessor):
    counter = stateful(0)  # Automatically saved and restored

    async def run(self):
        async for _ in self.run_loop(interval=1):
            self.counter += 1
            print(f"Count: {self.counter}")

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

With Resilience Features

from dory import DoryApp, BaseProcessor, stateful
from dory.resilience import CircuitBreaker, retry_with_backoff
from dory.monitoring import create_span

class MyApp(BaseProcessor):
    counter = stateful(0)
    db_breaker = CircuitBreaker(name="database", failure_threshold=5)

    @retry_with_backoff(max_attempts=3)
    async def fetch_data(self):
        with create_span("fetch_data"):
            return await self.db_breaker.call(self.database.query)

    async def run(self):
        async for _ in self.run_loop(interval=1):
            try:
                data = await self.fetch_data()
                self.counter += 1
            except Exception as e:
                self.context.logger().error(f"Failed: {e}")

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

Function-Based API

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
        ctx.logger().info(f"Count: {counter.value}")

Configuration

Zero-Config (Recommended)

No configuration file needed! The SDK auto-detects your environment:

Environment	Auto-Configuration
Kubernetes	Production preset, ConfigMap state, JSON logs, port 8080
Local	Development preset, local file state, colored logs, auto-port

Just run your code - the SDK handles everything.

Run multiple apps locally? Each auto-selects an available port (no conflicts).

Optional: dory.yaml

Only create dory.yaml if you need custom settings. DoryConfig recognizes exactly five keys (everything else is silently ignored via extra="ignore"):

startup_timeout_sec: 30        # 1-300, default 30
shutdown_timeout_sec: 30       # 1-300, default 30
health_port: 8080              # 0-65535, default 8080 (0 = auto-select)
state_backend: configmap       # configmap | pvc | s3 | local, default configmap
log_level: INFO                # DEBUG | INFO | WARNING | ERROR | CRITICAL, default INFO

Not supported in dory.yaml: retry, circuit-breaker, and OpenTelemetry tuning are not config-file keys. Any retry:, circuit_breaker:, or opentelemetry: block is silently dropped. Configure those components in code (see the resilience examples below) or via their constructor arguments in BaseProcessor.__init__.

See docs/08-configuration.md for the authoritative config reference.

Environment Variables

Variable	Default	Description
DORY_HEALTH_PORT	8080	Health server port
DORY_STATE_BACKEND	configmap	State storage (configmap/pvc/s3/local)
DORY_LOG_LEVEL	INFO	Log level
DORY_STARTUP_TIMEOUT_SEC	30	Startup timeout
DORY_SHUTDOWN_TIMEOUT_SEC	30	Shutdown timeout
DORY_STATE_TOKEN	—	Bearer token authenticating the /state endpoints (must match the orchestrator). Injected automatically on Dory.

See docs/08-configuration.md for the full runtime env var catalog.

API Reference

BaseProcessor

class MyApp(BaseProcessor):
    # Stateful fields (auto-saved/restored)
    counter = stateful(0)
    data = stateful(dict)

    async def startup(self):
        """Called once on startup (optional)"""
        pass

    async def run(self):
        """Main processing loop (required)"""
        async for i in self.run_loop(interval=1):
            self.counter += 1

    async def shutdown(self):
        """Called on graceful shutdown (optional)"""
        pass

    # Fault handling hooks (optional)
    async def on_state_restore_failed(self, error: Exception):
        """Called when state restoration fails"""
        pass

    async def on_rapid_restart_detected(self, restart_count: int):
        """Called when rapid restart loop detected"""
        pass

    def reset_caches(self):
        """Called on golden image reset"""
        pass

Circuit Breaker

from dory.resilience import CircuitBreaker, CircuitState

# Create circuit breaker
breaker = CircuitBreaker(
    name="database",
    failure_threshold=5,    # Open after 5 failures
    success_threshold=2,    # Close after 2 successes in half-open
    timeout_seconds=30.0,   # Seconds before trying half-open
)

# Use with async call
result = await breaker.call(async_function, arg1, arg2)

# Check state
if breaker.state == CircuitState.OPEN:
    print("Circuit is open, requests will fail fast")

# Manual control (both are async coroutines)
await breaker.open()   # Force open
await breaker.reset()  # Force closed

Retry with Backoff

from dory.resilience import retry_with_backoff, RetryPolicy

# Decorator usage
@retry_with_backoff(max_attempts=3, initial_delay=0.1)
async def flaky_operation():
    return await api.call()

# With custom policy
policy = RetryPolicy(
    max_attempts=5,
    initial_delay=0.1,
    multiplier=2.0,
    max_delay=30.0,
    jitter=True
)

@retry_with_backoff(policy=policy)
async def custom_retry():
    pass

Error Classification

from dory.errors import ErrorClassifier, ErrorType

classifier = ErrorClassifier()

try:
    await operation()
except Exception as e:
    result = classifier.classify(e)

    # ClassificationResult fields: error_type, recommended_action,
    # retryable, severity, details
    if result.retryable:
        # Retry the operation
        await retry_operation()
    else:
        # Don't retry, log and alert
        logger.error(
            f"Non-retryable error: {e} "
            f"(action={result.recommended_action}, severity={result.severity})"
        )

OpenTelemetry

from dory.monitoring import create_span, add_span_attributes, trace_function

# Context manager
with create_span("database_query", {"table": "users"}):
    result = await db.query("SELECT * FROM users")

# Decorator
@trace_function("process_item")
async def process_item(item):
    add_span_attributes({"item_id": item.id})
    return await transform(item)

ExecutionContext

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

    # Logging
    ctx.logger().info("Processing started")

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

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

HTTP Endpoints

Endpoint	Method	Description
/health	GET	Liveness probe (is process alive?)
/ready	GET	Readiness probe (ready to serve?)
/metrics	GET	Prometheus metrics
/state	GET	Export current state (Bearer auth — DORY_STATE_TOKEN)
/state	POST	Import/restore state (Bearer auth — DORY_STATE_TOKEN)
/prestop	GET	PreStop hook handler
/	GET	Service info + SDK version

/state is fail-closed. Both /state methods require Authorization: Bearer <DORY_STATE_TOKEN>. If no token is configured they deny all requests (HTTP 401). On Dory the token is provisioned and injected for you.

CLI Tool

The CLI provides two commands: dory init and dory validate.

# Initialize a new project (creates main.py + Dockerfile)
dory init my-app
dory init my-app -o ./my-app -f          # custom output dir, overwrite existing

# Validate configuration (loads dory.yaml + env overrides and prints settings)
dory validate
dory validate -c path/to/dory.yaml

dory init flags: name (positional), -o/--output, -i/--image (accepted but currently unused), -f/--force. dory validate flags: -c/--config.

State Migration Flow

Pod Shutdown

1. Kubernetes sends SIGTERM / calls /prestop
2. SDK marks processor as not-ready
3. SDK saves state to ConfigMap
4. Your shutdown() is called
5. Pod terminates

Pod Startup

1. New pod starts
2. SDK checks for existing state in ConfigMap
3. Your startup() is called
4. SDK restores state (calls restore_state or sets @stateful fields)
5. SDK marks processor as ready
6. Your run() starts

License

Apache 2.0