Structum Bootstrap (structum-bootstrap)¶

# ❌ Assunzioni implicite
app.run()

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

# ✅ Validazione esplicita
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 con report dettagliato

# Se arriviamo qui, l'ambiente è GARANTITO valido
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 per validatori custom."""
    
    def validate(self, context: ValidationContext) -> None:
        """
        Esegue validazione.
        Registra risultati in context.
        """
        ...

context = ValidationContext()

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

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

# Verifica stato
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 con 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 con 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 per 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)

class SystemBootstrapper:
    """Main bootstrapper orchestrator."""
    
    def add_validator(self, validator: Validator) -> None:
        """Add a validator to the bootstrap sequence."""
        ...
    
    def run(self) -> ValidationResult:
        """
        Run all validators.
        Returns ValidationResult with success status.
        """
        ...
    
    def run_or_exit(self, exit_code: int = 1) -> None:
        """
        Run validators and exit if validation fails.
        Use in production for fail-fast behavior.
        """
        ...

class EnvValidator(Validator):
    def __init__(
        self,
        required: list[str] | dict[str, str] = None,
        optional: list[str] | dict[str, str] = None
    ):
        ...

class PythonVersionValidator(Validator):
    def __init__(
        self,
        min_version: str = None,
        max_version: str = None,
        allowed_versions: list[str] = None
    ):
        ...

class DirectoryValidator(Validator):
    def __init__(
        self,
        path: str,
        readable: bool = True,
        writable: bool = False,
        create_if_missing: bool = False
    ):
        ...

class DatabaseValidator(Validator):
    def __init__(self, url: str, timeout: float = 5.0):
        ...
    
    @classmethod
    def from_config(cls, key: str = "database.url") -> "DatabaseValidator":
        ...

class ValidationContext:
    """Container for validation results."""
    
    def add_check(self, name: str, success: bool, message: str) -> None:
        """Register a validation check result."""
        ...
    
    def has_failures(self) -> bool:
        """Check if any validation failed."""
        ...
    
    def get_report(self) -> str:
        """Get formatted text report."""
        ...
    
    @property
    def total_checks(self) -> int:
        ...
    
    @property
    def successful_checks(self) -> int:
        ...
    
    @property
    def failed_checks(self) -> int:
        ...