Safeline

A Python framework for building safety-critical operations with arbitrary stage pipelines, audit trails, locking, and approval workflows.

Why Safeline?

When building tools that perform critical operations (deployments, data migrations, infrastructure changes), you need:

• Structured workflows - Break complex operations into reviewable stages
• Audit trails - Know who did what, when, and why
• Approval gates - Require sign-off before dangerous actions
• Rollback support - Undo operations when things go wrong
• Environment awareness - Different rules for prod vs dev

Safeline provides the infrastructure so you can focus on your domain logic.

Installation

pip install sfln

# With optional dependencies
pip install sfln[cli]      # CLI support (click)
pip install sfln[yaml]     # YAML config (pyyaml)
pip install sfln[all]      # All optional deps

Quick Example

Safeline recommends a standard stage pattern for safety-critical operations:

plan -> validate -> stage -> apply -> verify

• plan - Analyze what will be done, calculate risk
• validate - Check preconditions, ensure operation is safe to proceed
• stage - Prepare resources, create backups (reversible)
• apply - Execute the actual change (requires approval in production)
• verify - Confirm the operation succeeded

from sfln import Operation, Stage, StageResult, Context, Safeline, MemoryStore

class DeployOperation(Operation):
    name = "deploy"
    stages = [
        Stage("plan"),
        Stage("validate"),
        Stage("stage", reversible=True),
        Stage("apply", needs_approval=True, reversible=True),
        Stage("verify"),
    ]

    def plan(self, context: Context) -> StageResult:
        # Analyze what will be deployed
        return StageResult(data={"targets": context.targets, "version": "2.0.0"})

    def validate(self, context: Context) -> StageResult:
        # Check preconditions
        return StageResult(data={"valid": True})

    def stage(self, context: Context) -> StageResult:
        # Create backup before changes
        backup_id = create_backup()
        return StageResult(
            data={"backup_id": backup_id},
            rollback_data={"backup_id": backup_id}
        )

    def apply(self, context: Context) -> StageResult:
        # Execute the deployment
        version = context.get_stage_result("plan")["version"]
        return StageResult(data={"deployed": version})

    def verify(self, context: Context) -> StageResult:
        # Confirm deployment succeeded
        return StageResult(data={"healthy": True})

    def rollback_stage(self, context: Context) -> StageResult:
        return StageResult(data={"cleaned": True})

    def rollback_apply(self, context: Context) -> StageResult:
        backup_id = context.get_stage_result("stage")["backup_id"]
        restore_from_backup(backup_id)
        return StageResult(data={"restored": True})

# Run it
sl = Safeline(store=MemoryStore())
result = sl.run(DeployOperation(targets=["app-server-1"]))
print(f"Success: {result.success}")

Custom Stages

The standard stages are a recommendation, not a requirement. Define whatever stages make sense for your operation:

# Database migration with custom stages
stages = [
    Stage("analyze"),
    Stage("backup_schema"),
    Stage("migrate", needs_approval=True),
    Stage("verify_data"),
    Stage("cleanup"),
]

# Simple file operation
stages = [
    Stage("scan"),
    Stage("execute"),
]

Key Features

Standard Stage Pattern

The recommended pattern for safety-critical operations:

stages = [
    Stage("plan"),                                    # Analyze
    Stage("validate"),                                # Check preconditions
    Stage("stage", reversible=True),                  # Prepare/backup
    Stage("apply", needs_approval=True, reversible=True),  # Execute
    Stage("verify"),                                  # Confirm success
]

Per-Stage Controls

Each stage can have its own requirements:

Stage(
    name="apply",
    required=True,           # Operation fails if this fails
    needs_approval=True,     # Must be approved first
    needs_lock=True,         # Exclusive lock during execution
    reversible=True,         # Can be rolled back
    timeout_seconds=300,     # Max execution time
)

Audit Trail

All operations automatically emit events:

sl = Safeline(
    store=MemoryStore(),
    auditor=AuditStoreClient(url="http://audit-store:8080"),
)

Events: operation_created, stage_started, stage_completed, approval_granted, lock_acquired, etc.

Environment-Aware Protection

from sfln import get_protection_level

level = get_protection_level("production")
# level.require_approval = True
# level.require_backup = True
# level.allow_force = False

Development

# Setup
make quickstart
source .venv/bin/activate

# Test
make test

# Code quality
make check
make format

License

MIT