Structum Bootstrap (structum-bootstrap)¶

# ❌ Implicit Assumptions
app.run()

# Runtime errors after deploy:
# - "DATABASE_URL not set"
# - "/var/log/app not writable"  
# - "Python 3.8 required, found 3.7"

# ✅ Explicit Validation
boot = SystemBootstrapper()
boot.add_validator(EnvValidator(required=["DATABASE_URL"]))
boot.add_validator(DirectoryValidator("/var/log/app", writable=True))
boot.add_validator(PythonVersionValidator(min_version="3.9"))

boot.run_or_exit()  # Fails FAST with detailed report

# If we reach here, the environment is GUARANTEED valid
app.run()

bootstrapper = SystemBootstrapper()
bootstrapper.add_validator(validator1)
bootstrapper.add_validator(validator2)

# Run validation
result = bootstrapper.run()

if not result.success:
    print(result.report)
    sys.exit(1)

from typing import Protocol

class Validator(Protocol):
    """Protocol for custom validators."""
    
    def validate(self, context: ValidationContext) -> None:
        """
        Executes validation.
        Records results in context.
        """
        ...

context = ValidationContext()

# Record check success
context.add_check("Database", success=True, message="Connected OK")

# Record check failure
context.add_check("Redis", success=False, message="Connection timeout")

# Check status
if context.has_failures():
    print(context.get_report())

from structum_lab.plugins.bootstrap import (
    SystemBootstrapper,
    EnvValidator,
    PythonVersionValidator
)

def main():
    # 1. Create bootstrapper
    boot = SystemBootstrapper()
    
    # 2. Add validators
    boot.add_validator(
        EnvValidator(required=["DATABASE_URL", "API_KEY"])
    )
    boot.add_validator(
        PythonVersionValidator(min_version="3.9")
    )
    
    # 3. Run validation
    boot.run_or_exit()  # Exits with code 1 if validation fails
    
    # 4. Start application (guaranteed valid environment)
    print("✅ Environment validated - starting application")
    start_application()

if __name__ == "__main__":
    main()

from structum_lab.plugins.bootstrap import EnvValidator

# Required vars
validator = EnvValidator(required=["DB_URL", "API_KEY"])

# Optional vars (warns if missing)
validator = EnvValidator(
    required=["DB_URL"],
    optional=["REDIS_URL", "CACHE_TTL"]
)

# Custom names for reporting
validator = EnvValidator(
    required={
        "DATABASE_URL": "Primary Database Connection",
        "REDIS_URL": "Cache Server Connection"
    }
)

from structum_lab.plugins.bootstrap import PythonVersionValidator

# Minimum version
validator = PythonVersionValidator(min_version="3.9")

# Exact version range
validator = PythonVersionValidator(
    min_version="3.9",
    max_version="3.11"
)

# Specific versions
validator = PythonVersionValidator(
    allowed_versions=["3.10", "3.11"]
)

from structum_lab.plugins.bootstrap import DirectoryValidator

# Check existence
validator = DirectoryValidator("/var/log/myapp")

# Check writable
validator = DirectoryValidator(
    "/var/log/myapp",
    writable=True
)

# Check readable
validator = DirectoryValidator(
    "/etc/myapp",
    readable=True
)

# Auto-create if missing
validator = DirectoryValidator(
    "/var/log/myapp",
    writable=True,
    create_if_missing=True
)

from structum_lab.plugins.bootstrap import ConfigValidator

# Basic validation
validator = ConfigValidator()

# Validate specific keys exist
validator = ConfigValidator(
    required_keys=["database.url", "app.name"]
)

# Validate config provider
validator = ConfigValidator(
    provider_class=DynaconfConfigProvider
)

from structum_lab.plugins.bootstrap import DatabaseValidator

# Basic connectivity check
validator = DatabaseValidator(
    url="postgresql://localhost/mydb"
)

# With timeout
validator = DatabaseValidator(
    url="postgresql://localhost/mydb",
    timeout=5.0  # seconds
)

# From config
validator = DatabaseValidator.from_config()

from structum_lab.plugins.bootstrap import FileValidator

# Check file exists
validator = FileValidator("/etc/myapp/config.toml")

# Check readable
validator = FileValidator(
    "/etc/myapp/config.toml",
    readable=True
)

# Check specific size constraints
validator = FileValidator(
    "/var/lib/myapp/data.db",
    max_size_mb=1000
)

from structum_lab.plugins.bootstrap import Validator, ValidationContext

class PortAvailableValidator(Validator):
    """Check if network port is available."""
    
    def __init__(self, port: int):
        self.port = port
    
    def validate(self, context: ValidationContext) -> None:
        import socket
        
        try:
            sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
            sock.bind(('localhost', self.port))
            sock.close()
            
            context.add_check(
                f"Port {self.port}",
                success=True,
                message=f"Available"
            )
        except OSError:
            context.add_check(
                f"Port {self.port}",
                success=False,
                message=f"Already in use"
            )

# Usage
boot.add_validator(PortAvailableValidator(8000))

from datetime import datetime, timedelta

class SSLCertificateValidator(Validator):
    """Validate SSL certificate is not expired."""
    
    def __init__(self, cert_path: str, warn_days: int = 30):
        self.cert_path = cert_path
        self.warn_days = warn_days
    
    def validate(self, context: ValidationContext) -> None:
        from pathlib import Path
        import ssl
        import OpenSSL
        
        cert_file = Path(self.cert_path)
        
        if not cert_file.exists():
            context.add_check(
                "SSL Certificate",
                success=False,
                message=f"File not found: {self.cert_path}"
            )
            return
        
        # Load certificate
        with open(cert_file, 'rb') as f:
            cert_data = f.read()
        
        cert = OpenSSL.crypto.load_certificate(
            OpenSSL.crypto.FILETYPE_PEM,
            cert_data
        )
        
        # Check expiration
        not_after = datetime.strptime(
            cert.get_notAfter().decode('ascii'),
            '%Y%m%d%H%M%SZ'
        )
        
        days_until_expiry = (not_after - datetime.utcnow()).days
        
        if days_until_expiry < 0:
            context.add_check(
                "SSL Certificate",
                success=False,
                message=f"Expired {abs(days_until_expiry)} days ago"
            )
        elif days_until_expiry < self.warn_days:
            context.add_check(
                "SSL Certificate",
                success=True,
                message=f"⚠️  Expires in {days_until_expiry} days (warning threshold: {self.warn_days})"
            )
        else:
            context.add_check(
                "SSL Certificate",
                success=True,
                message=f"Valid for {days_until_expiry} days"
            )

import asyncio

class AsyncDatabaseValidator(Validator):
    """Async database connectivity check."""
    
    def __init__(self, url: str):
        self.url = url
    
    def validate(self, context: ValidationContext) -> None:
        # Run async code in sync context
        result = asyncio.run(self._async_check())
        
        context.add_check(
            "Database (Async)",
            success=result["success"],
            message=result["message"]
        )
    
    async def _async_check(self) -> dict:
        try:
            import asyncpg
            
            conn = await asyncpg.connect(self.url, timeout=5.0)
            await conn.close()
            
            return {
                "success": True,
                "message": "Connected successfully"
            }
        except Exception as e:
            return {
                "success": False,
                "message": f"Connection failed: {e}"
            }

from structum_lab.plugins.bootstrap import ValidationContext

context = ValidationContext()

# Add checks
context.add_check("Service A", True, "OK")
context.add_check("Service B", False, "Failed to connect")

# Query status
print(f"Total checks: {context.total_checks}")
print(f"Successful: {context.successful_checks}")
print(f"Failed: {context.failed_checks}")

# Get detailed report
print(context.get_report())

# Check specific results
for check in context.checks:
    print(f"{check.name}: {check.success}")
    if not check.success:
        print(f"  Error: {check.message}")

# Export to JSON
import json
report_json = {
    "total": context.total_checks,
    "passed": context.successful_checks,
    "failed": context.failed_checks,
    "checks": [
        {
            "name": check.name,
            "success": check.success,
            "message": check.message
        }
        for check in context.checks
    ]
}
print(json.dumps(report_json, indent=2))

# Production mode: exit immediately on failure
boot = SystemBootstrapper()
boot.add_validator(EnvValidator(required=["DB_URL"]))

boot.run_or_exit()  # sys.exit(1) if validation fails

# This line only executes if validation passed
start_application()

# Development mode: collect all errors
boot = SystemBootstrapper()
boot.add_validator(EnvValidator(required=["DB_URL"]))
boot.add_validator(DirectoryValidator("/var/log"))

result = boot.run()

if not result.success:
    print("⚠️  Found validation issues:")
    print(result.report)
    
    # Developer can decide whether to continue
    if input("Continue anyway? (y/n): ").lower() != 'y':
        sys.exit(1)

start_application()

class CustomBootstrapper(SystemBootstrapper):
    """Custom bootstrapper with logging."""
    
    def on_validation_start(self):
        """Called before validation starts."""
        logger.info("Starting bootstrap validation")
    
    def on_validation_complete(self, context: ValidationContext):
        """Called after all validators run."""
        logger.info(
            f"Validation complete: {context.successful_checks}/{context.total_checks} passed"
        )
        
        if context.has_failures():
            logger.error(f"Validation failed:\n{context.get_report()}")
            
            # Send to monitoring
            metrics.increment("bootstrap.failures")
    
    def on_validator_error(self, validator: Validator, error: Exception):
        """Called if validator raises exception."""
        logger.exception(f"Validator {validator.__class__.__name__} crashed: {error}")

import os

boot = SystemBootstrapper()

# Always validate
boot.add_validator(PythonVersionValidator(min_version="3.9"))

# Production-only validators
if os.getenv("ENV") == "production":
    boot.add_validator(SSLCertificateValidator("/etc/ssl/app.pem"))
    boot.add_validator(DirectoryValidator("/var/log", writable=True))

# Development-only validators
if os.getenv("ENV") == "development":
    boot.add_validator(PortAvailableValidator(8000))

boot.run_or_exit()

class InfrastructureValidators:
    """Group of infrastructure validators."""
    
    @staticmethod
    def get_all() -> list[Validator]:
        return [
            EnvValidator(required=["DB_URL", "REDIS_URL"]),
            DatabaseValidator.from_config(),
            PortAvailableValidator(5432),  # PostgreSQL
            PortAvailableValidator(6379),  # Redis
        ]

class SecurityValidators:
    """Group of security validators."""
    
    @staticmethod
    def get_all() -> list[Validator]:
        return [
            SSLCertificateValidator("/etc/ssl/app.pem"),
            FileValidator("/etc/ssl/private/app.key", readable=True),
            EnvValidator(required=["JWT_SECRET", "ENCRYPTION_KEY"]),
        ]

# Usage
boot = SystemBootstrapper()

for validator in InfrastructureValidators.get_all():
    boot.add_validator(validator)

for validator in SecurityValidators.get_all():
    boot.add_validator(validator)

boot.run_or_exit()

def bootstrap_application():
    """Multi-stage bootstrap with fail-fast."""
    
    # Stage 1: Critical checks (fail immediately)
    stage1 = SystemBootstrapper()
    stage1.add_validator(PythonVersionValidator(min_version="3.9"))
    stage1.add_validator(EnvValidator(required=["DATABASE_URL"]))
    stage1.run_or_exit()
    
    print("✓ Stage 1: Critical checks passed")
    
    # Stage 2: Infrastructure (can warn)
    stage2 = SystemBootstrapper()
    stage2.add_validator(DatabaseValidator.from_config())
    stage2.add_validator(DirectoryValidator("/var/log", create_if_missing=True))
    
    result = stage2.run()
    if not result.success:
        print(f"⚠️  Stage 2: Infrastructure warnings:\n{result.report}")
    
    print("✓ Stage 2: Infrastructure checks completed")
    
    # Stage 3: Optional features
    stage3 = SystemBootstrapper()
    stage3.add_validator(EnvValidator(optional=["REDIS_URL", "CACHE_TTL"]))
    stage3.run()  # Never fails
    
    print("✓ Stage 3: Optional features validated")
    print("✅ Bootstrap complete - starting application")

# health_check.py
from structum_lab.plugins.bootstrap import SystemBootstrapper, DatabaseValidator

def health_check() -> bool:
    """Health check for Kubernetes/Docker."""
    boot = SystemBootstrapper()
    boot.add_validator(DatabaseValidator.from_config())
    
    result = boot.run()
    return result.success

if __name__ == "__main__":
    import sys
    sys.exit(0 if health_check() else 1)