Session 10: Production System Integration & Deployment - Welcome to the Big Leagues¶

from typing import Dict, Any, List, Protocol
from dataclasses import dataclass, field
from enum import Enum
import asyncio
import aiohttp
import logging
from datetime import datetime, timedelta

class IntegrationPattern(Enum):
    """Common production integration patterns."""
    REQUEST_REPLY = "request_reply"
    PUBLISH_SUBSCRIBE = "publish_subscribe"
    MESSAGE_QUEUE = "message_queue"
    API_GATEWAY = "api_gateway"
    SERVICE_MESH = "service_mesh"

@dataclass
class SystemConnection:
    """Configuration for production system connections."""
    system_name: str
    endpoint: str
    auth_method: str
    credentials: Dict[str, Any]
    retry_policy: Dict[str, int] = field(default_factory=dict)
    circuit_breaker: bool = True
    timeout_seconds: int = 30
    rate_limit: Optional[int] = None

# Standard adapter interface for production systems

class ProductionSystemAdapter(Protocol):
    """Protocol for production system adapters."""

    async def connect(self) -> bool:
        """Establish connection to production system."""
        ...

    async def authenticate(self, credentials: Dict[str, Any]) -> bool:
        """Authenticate with production system."""
        ...

    async def execute_operation(self, operation: str,
                               params: Dict[str, Any]) -> Any:
        """Execute operation on production system."""
        ...

    async def health_check(self) -> bool:
        """Check system health and connectivity."""
        ...

class DataSystemIntegrationAdapter:
    """Adapter for distributed data system integration."""

    def __init__(self, base_url: str, credentials: Dict[str, Any]):
        self.base_url = base_url.rstrip('/')
        self.credentials = credentials
        self.session: Optional[aiohttp.ClientSession] = None
        self.auth_token: Optional[str] = None
        self.token_expires: Optional[datetime] = None
        self.logger = logging.getLogger(__name__)

    async def connect(self) -> bool:
        """Establish connection to data system with enterprise configuration."""
        try:
            # Configure TCP connector with enterprise settings
            connector = aiohttp.TCPConnector(
                limit=50,  # Total connection pool size
                limit_per_host=10,  # Connections per host
                keepalive_timeout=30  # Keep connections alive
            )

            # HTTP session configuration
            self.session = aiohttp.ClientSession(
                connector=connector,
                timeout=aiohttp.ClientTimeout(total=30),
                headers={
                    "User-Agent": "EnterpriseAgent-DataSystem/1.0",
                    "Accept": "application/json"
                }
            )
            return await self.authenticate(self.credentials)
        except Exception as e:
            self.logger.error(f"Data system connection failed: {e}")
            return False

    async def authenticate(self, credentials: Dict[str, Any]) -> bool:
        """Authenticate with data system using OAuth 2.0."""
        auth_url = f"{self.base_url}/oauth/token"

        # Prepare OAuth 2.0 client credentials grant
        auth_data = {
            "grant_type": "client_credentials",
            "client_id": credentials["client_id"],
            "client_secret": credentials["client_secret"],
            "scope": credentials.get("scope", "read write")
        }

        try:
            async with self.session.post(auth_url, data=auth_data) as response:
                if response.status == 200:
                    token_data = await response.json()
                    self.auth_token = token_data["access_token"]
                    expires_in = token_data.get("expires_in", 3600)
                    self.token_expires = datetime.now() + timedelta(seconds=expires_in)

                    # Update session headers with bearer token
                    self.session.headers.update({
                        "Authorization": f"Bearer {self.auth_token}"
                    })
                    return True
                else:
                    self.logger.error(f"Data system auth failed: {response.status}")
                    return False
        except Exception as e:
            self.logger.error(f"Data system authentication error: {e}")
            return False

    async def get_dataset(self, dataset_id: str) -> Dict[str, Any]:
        """Retrieve dataset from data system."""
        if not dataset_id:
            raise ValueError("dataset_id is required")

        # Ensure we have valid authentication
        if not await self._ensure_authenticated():
            raise Exception("Authentication failed")

        # Build data system API URL
        service_url = f"{self.base_url}/api/v1/datasets"
        entity_url = f"{service_url}/{dataset_id}"

        try:
            async with self.session.get(entity_url) as response:
                if response.status == 200:
                    data = await response.json()
                    return self._transform_dataset(data)
                elif response.status == 404:
                    return {"error": "Dataset not found", "dataset_id": dataset_id}
                else:
                    error_text = await response.text()
                    raise Exception(f"Data system API error: {response.status} - {error_text}")
        except Exception as e:
            self.logger.error(f"Dataset retrieval failed: {e}")
            raise

    async def _ensure_authenticated(self) -> bool:
        """Ensure we have a valid authentication token."""
        if not self.auth_token:
            return await self.authenticate(self.credentials)

        # Check if token is about to expire (refresh 5 minutes early)
        if self.token_expires:
            time_until_expiry = self.token_expires - datetime.now()
            if time_until_expiry.total_seconds() < 300:  # 5 minutes
                self.logger.info("Token expiring soon, refreshing...")
                return await self.authenticate(self.credentials)

        return True

from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine
from sqlalchemy.orm import sessionmaker

class EnterpriseDatabase:
    """Enterprise database connection manager."""

    def __init__(self, database_configs: Dict[str, Dict[str, Any]]):
        self.configs = database_configs
        self.engines = {}
        self.session_factories = {}

    async def initialize_connections(self):
        """Initialize all configured database connections."""
        for db_name, config in self.configs.items():
            await self._setup_database(db_name, config)

    async def _setup_postgresql(self, db_name: str, config: Dict[str, Any]):
        """Setup PostgreSQL connection with enterprise configuration."""
        connection_string = (
            f"postgresql+asyncpg://{config['username']}:{config['password']}"
            f"@{config['host']}:{config['port']}/{config['database']}"
        )

        # Database engine configuration
        engine = create_async_engine(
            connection_string,
            pool_size=config.get("pool_size", 10),
            max_overflow=config.get("max_overflow", 20),
            pool_pre_ping=True,  # Validate connections before use
            pool_recycle=3600,   # Recycle connections every hour
            echo=config.get("debug", False)
        )

        # Session factory setup
        self.engines[db_name] = engine
        self.session_factories[db_name] = sessionmaker(
            bind=engine,
            class_=AsyncSession,
            expire_on_commit=False
        )

from typing import Dict, List, Optional, Set
from datetime import datetime, timedelta
import jwt
import bcrypt
from enum import Enum
from dataclasses import dataclass

class AuthenticationMethod(Enum):
    """Supported authentication methods."""
    BASIC = "basic"
    OAUTH2 = "oauth2"
    SAML = "saml"
    LDAP = "ldap"
    MFA = "mfa"
    CERTIFICATE = "certificate"

@dataclass
class UserContext:
    """User context with comprehensive attributes."""
    user_id: str
    username: str
    email: str
    roles: Set[str]
    permissions: Set[str]
    department: str
    security_clearance: str
    mfa_verified: bool = False
    last_authenticated: Optional[datetime] = None
    session_expires: Optional[datetime] = None

class EnterpriseAuthManager:
    """Comprehensive authentication and authorization manager."""

    def __init__(self, config: Dict[str, Any]):
        self.config = config
        self.active_sessions: Dict[str, UserContext] = {}
        self.failed_attempts: Dict[str, int] = {}
        self.blacklisted_tokens: Set[str] = set()

    async def authenticate(self, credentials: Dict[str, Any]) -> Optional[UserContext]:
        """Authenticate user with multiple factor support."""
        auth_method = credentials.get("method", "basic")
        username = credentials.get("username")

        # Check for account lockout
        if self._is_account_locked(username):
            raise AuthenticationError("Account locked due to failed attempts")

        try:
            if auth_method == AuthenticationMethod.BASIC.value:
                return await self._authenticate_basic(credentials)
            elif auth_method == AuthenticationMethod.OAUTH2.value:
                return await self._authenticate_oauth2(credentials)
            elif auth_method == AuthenticationMethod.LDAP.value:
                return await self._authenticate_ldap(credentials)
            else:
                raise AuthenticationError(f"Unsupported auth method: {auth_method}")
        except AuthenticationError:
            self._record_failed_attempt(username)
            raise

    def _is_account_locked(self, username: str) -> bool:
        """Check if account is locked due to failed attempts."""
        attempts = self.failed_attempts.get(username, 0)
        return attempts >= 5  # Lock after 5 failed attempts

    def _record_failed_attempt(self, username: str):
        """Record failed authentication attempt."""
        self.failed_attempts[username] = self.failed_attempts.get(username, 0) + 1

from enum import Enum

class Permission(Enum):
    """System permissions with hierarchical structure."""
    # Agent operations
    AGENT_CREATE = "agent:create"
    AGENT_READ = "agent:read"
    AGENT_UPDATE = "agent:update"
    AGENT_DELETE = "agent:delete"
    AGENT_EXECUTE = "agent:execute"

    # Data access by classification
    DATA_READ_PUBLIC = "data:read:public"
    DATA_READ_INTERNAL = "data:read:internal"
    DATA_READ_CONFIDENTIAL = "data:read:confidential"
    DATA_READ_SECRET = "data:read:secret"

    # System administration
    SYSTEM_CONFIG = "system:config"
    SYSTEM_MONITOR = "system:monitor"
    SYSTEM_ADMIN = "system:admin"

@dataclass
class Role:
    """Role definition with permissions and constraints."""
    name: str
    permissions: Set[Permission]
    constraints: Dict[str, Any] = field(default_factory=dict)
    inherits_from: List[str] = field(default_factory=list)

class RBACManager:
    """Role-Based Access Control manager."""

    def __init__(self):
        self.roles: Dict[str, Role] = {}
        self.user_roles: Dict[str, Set[str]] = {}
        self._initialize_default_roles()

    def _initialize_default_roles(self):
        """Initialize standard enterprise roles."""
        # Data Analyst role
        self.roles["data_analyst"] = Role(
            name="data_analyst",
            permissions={
                Permission.AGENT_READ,
                Permission.AGENT_EXECUTE,
                Permission.DATA_READ_PUBLIC,
                Permission.DATA_READ_INTERNAL
            },
            constraints={
                "time_restriction": "business_hours",
                "ip_whitelist": ["10.0.0.0/8"],
                "max_concurrent_sessions": 3
            }
        )

        # System Administrator role
        self.roles["system_admin"] = Role(
            name="system_admin",
            permissions={
                Permission.AGENT_CREATE,
                Permission.AGENT_READ,
                Permission.AGENT_UPDATE,
                Permission.AGENT_DELETE,
                Permission.SYSTEM_CONFIG,
                Permission.SYSTEM_MONITOR,
                Permission.SYSTEM_ADMIN
            }
        )

from cryptography.fernet import Fernet, MultiFernet
import os
import base64
import json

class EnterpriseEncryption:
    """Enterprise-grade encryption service."""

    def __init__(self, config: Dict[str, Any]):
        self.config = config
        self.symmetric_keys = self._initialize_symmetric_keys()
        self.key_rotation_interval = config.get("key_rotation_hours", 24)

    def _initialize_symmetric_keys(self) -> MultiFernet:
        """Initialize symmetric encryption with key rotation support."""
        keys = []

        # Primary key (current)
        primary_key = os.environ.get("ENCRYPTION_KEY_PRIMARY")
        if primary_key:
            keys.append(Fernet(primary_key.encode()))
        else:
            # Generate new key if none provided
            keys.append(Fernet(Fernet.generate_key()))

        # Secondary key (for rotation)
        secondary_key = os.environ.get("ENCRYPTION_KEY_SECONDARY")
        if secondary_key:
            keys.append(Fernet(secondary_key.encode()))

        return MultiFernet(keys)

    async def encrypt_sensitive_data(self, data: Dict[str, Any],
                                   classification: str = "internal") -> Dict[str, Any]:
        """Encrypt data based on classification level."""
        encrypted_data = {}

        for key, value in data.items():
            field_classification = self._get_field_classification(key, classification)

            if field_classification in ["confidential", "secret"]:
                # Use asymmetric encryption for highly sensitive data
                encrypted_value = self._asymmetric_encrypt(str(value))
                encrypted_data[key] = {
                    "value": encrypted_value,
                    "encrypted": True,
                    "method": "asymmetric",
                    "classification": field_classification
                }
            elif field_classification == "internal":
                # Use symmetric encryption for internal data
                encrypted_value = self.symmetric_keys.encrypt(str(value).encode())
                encrypted_data[key] = {
                    "value": base64.b64encode(encrypted_value).decode(),
                    "encrypted": True,
                    "method": "symmetric",
                    "classification": field_classification
                }
            else:
                # Public data doesn't need encryption
                encrypted_data[key] = {
                    "value": value,
                    "encrypted": False,
                    "classification": field_classification
                }

        return encrypted_data

from typing import Dict, Any, List
from dataclasses import dataclass, field
from datetime import datetime
from enum import Enum

class MetricType(Enum):
    """Types of metrics collected."""
    COUNTER = "counter"      # Monotonically increasing values
    GAUGE = "gauge"          # Snapshot values that can go up or down
    HISTOGRAM = "histogram"  # Distribution of values with buckets
    SUMMARY = "summary"      # Distribution with quantiles

@dataclass
class Metric:
    """Individual metric definition."""
    name: str
    type: MetricType
    value: float
    labels: Dict[str, str] = field(default_factory=dict)
    timestamp: datetime = field(default_factory=datetime.now)
    description: str = ""

class BasicMonitoring:
    """Basic monitoring system for production agents."""

    def __init__(self, config: Dict[str, Any]):
        self.config = config
        self.metrics_store: Dict[str, List[Metric]] = {}

    async def collect_agent_metrics(self, agent_id: str) -> Dict[str, Any]:
        """Collect basic agent performance metrics."""
        metrics = {
            "agent_id": agent_id,
            "timestamp": datetime.now().isoformat(),
            "performance": await self._collect_performance_metrics(agent_id),
            "health": await self._collect_health_metrics(agent_id),
            "errors": await self._collect_error_metrics(agent_id)
        }

        return metrics

    async def _collect_performance_metrics(self, agent_id: str) -> Dict[str, Any]:
        """Collect agent performance metrics."""
        return {
            "cpu_usage_percent": 0.0,  # Implement actual collection
            "memory_usage_mb": 0.0,
            "request_latency_ms": 0.0,
            "throughput_rps": 0.0,
            "error_rate_percent": 0.0
        }

class HealthChecker:
    """Production health monitoring system."""

    def __init__(self, system_components: Dict[str, Any]):
        self.components = system_components

    async def comprehensive_health_check(self) -> Dict[str, Any]:
        """Perform comprehensive system health check."""
        health_status = {
            "overall_status": "unknown",
            "timestamp": datetime.now().isoformat(),
            "components": {}
        }

        all_healthy = True

        # Check database connectivity
        db_health = await self._check_database_health()
        health_status["components"]["database"] = db_health
        if not db_health["healthy"]:
            all_healthy = False

        # Check external API connectivity
        api_health = await self._check_api_health()
        health_status["components"]["external_apis"] = api_health
        if not api_health["healthy"]:
            all_healthy = False

        # Check system resources
        resource_health = await self._check_system_resources()
        health_status["components"]["resources"] = resource_health
        if not resource_health["healthy"]:
            all_healthy = False

        health_status["overall_status"] = "healthy" if all_healthy else "unhealthy"
        return health_status

    async def _check_database_health(self) -> Dict[str, Any]:
        """Check database connectivity and performance."""
        try:
            # Implement actual database health check
            return {"healthy": True, "response_time_ms": 10}
        except Exception as e:
            return {"healthy": False, "error": str(e)}

from enum import Enum

class AlertSeverity(Enum):
    """Alert severity levels."""
    INFO = "info"
    WARNING = "warning"
    ERROR = "error"
    CRITICAL = "critical"

@dataclass
class Alert:
    """Basic alert definition."""
    alert_id: str
    name: str
    description: str
    severity: AlertSeverity
    triggered_at: datetime
    resolved_at: Optional[datetime] = None

class BasicAlertManager:
    """Basic alert management system."""

    def __init__(self, config: Dict[str, Any]):
        self.config = config
        self.active_alerts: Dict[str, Alert] = {}
        self.alert_thresholds = {
            "cpu_usage": 80.0,
            "memory_usage": 85.0,
            "error_rate": 5.0,
            "response_time": 2000.0  # milliseconds
        }

    async def evaluate_alerts(self, metrics: Dict[str, Any]):
        """Evaluate metrics against alert thresholds."""
        performance = metrics.get("performance", {})

        # Check CPU usage
        cpu_usage = performance.get("cpu_usage_percent", 0)
        if cpu_usage > self.alert_thresholds["cpu_usage"]:
            await self._create_alert(
                "high_cpu_usage",
                f"High CPU usage: {cpu_usage}%",
                AlertSeverity.WARNING
            )

    async def _create_alert(self, alert_name: str, description: str,
                           severity: AlertSeverity):
        """Create and store alert."""
        alert_id = f"{alert_name}_{datetime.now().strftime('%Y%m%d_%H%M%S')}"

        alert = Alert(
            alert_id=alert_id,
            name=alert_name,
            description=description,
            severity=severity,
            triggered_at=datetime.now()
        )

        self.active_alerts[alert_id] = alert
        await self._send_notification(alert)

    async def _send_notification(self, alert: Alert):
        """Send alert notification (implement based on your needs)."""
        print(f"ALERT: {alert.severity.value.upper()} - {alert.description}")