SOLID Principles in Modern Software Architecture

# ❌ Violation: UserService doing too much
class UserService:
    def create_user(self, user_data):
        # Validate user data
        if not self._validate_email(user_data['email']):
            raise ValidationError("Invalid email")

        # Save to database
        user = self.db.save_user(user_data)

        # Send welcome email
        self.email_service.send_welcome_email(user.email)

        # Log analytics event
        self.analytics.track_user_registration(user.id)

        # Update recommendation engine
        self.recommendations.initialize_user_profile(user.id)

        return user

# ✅ SRP Compliant: Focused responsibilities with comprehensive error handling
class UserRegistrationOrchestrator:
    def __init__(self, user_repository, notification_service, 
                 analytics_service, recommendation_service, logger):
        self.user_repository = user_repository
        self.notification_service = notification_service
        self.analytics_service = analytics_service
        self.recommendation_service = recommendation_service
        self.logger = logger

    async def register_user(self, user_data):
        try:
            # Validate and create user with retry logic
            user = await self._create_user_with_retry(user_data)

            # Trigger downstream operations asynchronously with circuit breaker
            tasks = [
                self._safe_execute(
                    self.notification_service.send_welcome_email(user.email),
                    "notification"
                ),
                self._safe_execute(
                    self.analytics_service.track_registration(user.id),
                    "analytics"
                ),
                self._safe_execute(
                    self.recommendation_service.initialize_profile(user.id),
                    "recommendations"
                )
            ]

            # Execute with timeout and error isolation
            await asyncio.wait_for(
                asyncio.gather(*tasks, return_exceptions=True),
                timeout=10.0
            )

            return user

        except ValidationError as e:
            self.logger.warning(f"User registration validation failed: {e}")
            raise
        except DatabaseError as e:
            self.logger.error(f"Database error during registration: {e}")
            raise
        except asyncio.TimeoutError:
            self.logger.error("Registration timeout - downstream services slow")
            # User created successfully, downstream failures are non-critical
            return user

    async def _create_user_with_retry(self, user_data, max_retries=3):
        for attempt in range(max_retries):
            try:
                validated_data = await self.user_repository.validate(user_data)
                return await self.user_repository.create(validated_data)
            except TransientError as e:
                if attempt == max_retries - 1:
                    raise
                await asyncio.sleep(2 ** attempt)  # Exponential backoff

    async def _safe_execute(self, coroutine, service_name):
        try:
            return await coroutine
        except Exception as e:
            self.logger.error(f"{service_name} service error: {e}")
            # Return None or default value to prevent cascade failures
            return None

class UserRepository:
    def __init__(self, database, validator, metrics):
        self.database = database
        self.validator = validator
        self.metrics = metrics

    async def create(self, user_data):
        start_time = time.time()
        try:
            validated_data = await self.validator.validate(user_data)
            user = await self.database.users.create(validated_data)

            # Record success metrics
            self.metrics.increment('user.registration.success')
            self.metrics.histogram('user.registration.duration', 
                                 time.time() - start_time)
            return user

        except ValidationError:
            self.metrics.increment('user.registration.validation_error')
            raise
        except DatabaseError:
            self.metrics.increment('user.registration.database_error')
            raise

    async def validate(self, user_data):
        # Single responsibility: user data validation and business rules
        required_fields = ['email', 'name', 'password']
        for field in required_fields:
            if not user_data.get(field):
                raise ValidationError(f"Missing required field: {field}")

        # Email uniqueness check
        existing_user = await self.database.users.find_by_email(
            user_data['email']
        )
        if existing_user:
            raise ValidationError("Email already registered")

        # Password strength validation
        if len(user_data['password']) < 8:
            raise ValidationError("Password must be at least 8 characters")

        return user_data

# Base abstraction for payment processing
from abc import ABC, abstractmethod
from typing import Dict, Any
from dataclasses import dataclass

@dataclass
class PaymentRequest:
    amount: float
    currency: str
    customer_id: str
    metadata: Dict[str, Any]

@dataclass
class PaymentResult:
    success: bool
    transaction_id: str
    error_message: str = None

class PaymentProcessor(ABC):
    @abstractmethod
    async def process_payment(self, request: PaymentRequest) -> PaymentResult:
        pass

    @abstractmethod
    def supports_currency(self, currency: str) -> bool:
        pass

# Existing implementation - never modified
class StripePaymentProcessor(PaymentProcessor):
    def __init__(self, api_key: str):
        self.stripe_client = StripeClient(api_key)

    async def process_payment(self, request: PaymentRequest) -> PaymentResult:
        try:
            result = await self.stripe_client.charge(
                amount=request.amount,
                currency=request.currency,
                customer=request.customer_id
            )
            return PaymentResult(
                success=True,
                transaction_id=result.id
            )
        except StripeError as e:
            return PaymentResult(
                success=False,
                transaction_id="",
                error_message=str(e)
            )

    def supports_currency(self, currency: str) -> bool:
        return currency in ['USD', 'EUR', 'GBP']

# New extension - no existing code modified
class PayPalPaymentProcessor(PaymentProcessor):
    def __init__(self, client_id: str, client_secret: str):
        self.paypal_client = PayPalClient(client_id, client_secret)

    async def process_payment(self, request: PaymentRequest) -> PaymentResult:
        # PayPal-specific implementation
        pass

    def supports_currency(self, currency: str) -> bool:
        return currency in ['USD', 'EUR', 'JPY', 'CAD']

# Payment orchestrator that leverages OCP
class PaymentOrchestrator:
    def __init__(self):
        self.processors: List[PaymentProcessor] = []

    def register_processor(self, processor: PaymentProcessor):
        self.processors.append(processor)

    async def process_payment(self, request: PaymentRequest) -> PaymentResult:
        for processor in self.processors:
            if processor.supports_currency(request.currency):
                result = await processor.process_payment(request)
                if result.success:
                    return result

        return PaymentResult(
            success=False,
            transaction_id="",
            error_message="No suitable payment processor found"
        )

# payment-config.yaml
payment_processors:
  - type: stripe
    enabled: true
    config:
      api_key: ${STRIPE_API_KEY}
      supported_currencies: ["USD", "EUR", "GBP"]

  - type: paypal
    enabled: true
    config:
      client_id: ${PAYPAL_CLIENT_ID}
      client_secret: ${PAYPAL_CLIENT_SECRET}
      supported_currencies: ["USD", "EUR", "JPY", "CAD"]

  - type: cryptocurrency
    enabled: false
    config:
      bitcoin_address: ${BTC_ADDRESS}
      supported_currencies: ["BTC", "ETH"]

# ❌ LSP Violation: Inconsistent behavior
class CacheInterface(ABC):
    @abstractmethod
    async def get(self, key: str) -> Optional[str]:
        pass

    @abstractmethod
    async def set(self, key: str, value: str, ttl: int = 3600) -> bool:
        pass

class RedisCache(CacheInterface):
    async def get(self, key: str) -> Optional[str]:
        return await self.redis_client.get(key)

    async def set(self, key: str, value: str, ttl: int = 3600) -> bool:
        result = await self.redis_client.setex(key, ttl, value)
        return result == "OK"

class MemoryCache(CacheInterface):
    async def get(self, key: str) -> Optional[str]:
        # ❌ Violation: Behavior differs from contract
        if key.startswith("temp_"):
            return None  # Arbitrary filtering not in contract
        return self.memory_store.get(key)

    async def set(self, key: str, value: str, ttl: int = 3600) -> bool:
        # ❌ Violation: Ignores TTL parameter
        self.memory_store[key] = value
        return True

# ✅ LSP Compliant: Consistent behavior with security considerations
import time
import asyncio
import hashlib
from abc import ABC, abstractmethod
from typing import Optional, Dict, Tuple
from cryptography.fernet import Fernet

class CacheInterface(ABC):
    @abstractmethod
    async def get(self, key: str) -> Optional[str]:
        """
        Retrieve value by key. Returns None if not found or expired.

        Security: Keys should be validated to prevent injection attacks.
        Performance: Implementation must handle concurrent access safely.
        """
        pass

    @abstractmethod
    async def set(self, key: str, value: str, ttl: int = 3600) -> bool:
        """
        Store value with TTL in seconds. Returns True if successful.

        Security: Values should be encrypted for sensitive data.
        Performance: Must be atomic operation.
        """
        pass

    @abstractmethod
    async def delete(self, key: str) -> bool:
        """Remove key from cache. Returns True if key existed."""
        pass

    @abstractmethod
    async def get_stats(self) -> Dict[str, int]:
        """Return cache statistics for monitoring."""
        pass

class SecureMemoryCache(CacheInterface):
    def __init__(self, encryption_key: bytes = None, max_size: int = 10000):
        self.store: Dict[str, Tuple[str, float]] = {}
        self.access_count: Dict[str, int] = {}
        self.max_size = max_size
        self.encryption_key = encryption_key
        self.cipher = Fernet(encryption_key) if encryption_key else None
        self._lock = asyncio.Lock()

        # Security: Track access patterns
        self.stats = {
            'hits': 0,
            'misses': 0,
            'sets': 0,
            'deletes': 0,
            'evictions': 0
        }

    def _validate_key(self, key: str) -> str:
        """Security: Prevent key injection and ensure safe storage."""
        if not key or len(key) > 250:
            raise ValueError("Invalid key: must be 1-250 characters")

        # Hash key to prevent path traversal and ensure consistent length
        return hashlib.sha256(key.encode()).hexdigest()

    def _encrypt_value(self, value: str) -> str:
        """Encrypt sensitive values if encryption is enabled."""
        if self.cipher:
            return self.cipher.encrypt(value.encode()).decode()
        return value

    def _decrypt_value(self, encrypted_value: str) -> str:
        """Decrypt values if encryption is enabled."""
        if self.cipher:
            return self.cipher.decrypt(encrypted_value.encode()).decode()
        return encrypted_value

    async def get(self, key: str) -> Optional[str]:
        safe_key = self._validate_key(key)

        async with self._lock:
            if safe_key in self.store:
                value, expiry = self.store[safe_key]

                if time.time() < expiry:
                    self.access_count[safe_key] = self.access_count.get(safe_key, 0) + 1
                    self.stats['hits'] += 1
                    return self._decrypt_value(value)
                else:
                    # Clean expired key
                    del self.store[safe_key]
                    self.access_count.pop(safe_key, None)
                    self.stats['evictions'] += 1

            self.stats['misses'] += 1
            return None

    async def set(self, key: str, value: str, ttl: int = 3600) -> bool:
        safe_key = self._validate_key(key)
        encrypted_value = self._encrypt_value(value)
        expiry = time.time() + ttl

        async with self._lock:
            # Implement LRU eviction if cache is full
            if len(self.store) >= self.max_size and safe_key not in self.store:
                await self._evict_lru()

            self.store[safe_key] = (encrypted_value, expiry)
            self.stats['sets'] += 1
            return True

    async def delete(self, key: str) -> bool:
        safe_key = self._validate_key(key)

        async with self._lock:
            if safe_key in self.store:
                del self.store[safe_key]
                self.access_count.pop(safe_key, None)
                self.stats['deletes'] += 1
                return True
            return False

    async def get_stats(self) -> Dict[str, int]:
        async with self._lock:
            return {
                **self.stats,
                'size': len(self.store),
                'max_size': self.max_size
            }

    async def _evict_lru(self):
        """Remove least recently used item."""
        if self.access_count:
            lru_key = min(self.access_count.keys(), 
                         key=lambda k: self.access_count[k])
            del self.store[lru_key]
            del self.access_count[lru_key]
            self.stats['evictions'] += 1

class RedisCache(CacheInterface):
    def __init__(self, redis_client, key_prefix: str = "app:"):
        self.redis_client = redis_client
        self.key_prefix = key_prefix

    def _make_key(self, key: str) -> str:
        """Create namespaced key with validation."""
        if not key or len(key) > 200:
            raise ValueError("Invalid key length")
        return f"{self.key_prefix}{key}"

    async def get(self, key: str) -> Optional[str]:
        safe_key = self._make_key(key)
        try:
            value = await self.redis_client.get(safe_key)
            return value.decode() if value else None
        except Exception as e:
            # Log error but don't break application flow
            logger.error(f"Redis get error for key {safe_key}: {e}")
            return None

    async def set(self, key: str, value: str, ttl: int = 3600) -> bool:
        safe_key = self._make_key(key)
        try:
            result = await self.redis_client.setex(safe_key, ttl, value)
            return result == "OK"
        except Exception as e:
            logger.error(f"Redis set error for key {safe_key}: {e}")
            return False

    async def delete(self, key: str) -> bool:
        safe_key = self._make_key(key)
        try:
            result = await self.redis_client.delete(safe_key)
            return result > 0
        except Exception as e:
            logger.error(f"Redis delete error for key {safe_key}: {e}")
            return False

    async def get_stats(self) -> Dict[str, int]:
        try:
            info = await self.redis_client.info()
            return {
                'hits': info.get('keyspace_hits', 0),
                'misses': info.get('keyspace_misses', 0),
                'size': info.get('db0', {}).get('keys', 0),
                'memory_usage': info.get('used_memory', 0)
            }
        except Exception as e:
            logger.error(f"Redis stats error: {e}")
            return {}

# Client code works consistently with any implementation
class UserService:
    def __init__(self, cache: CacheInterface, circuit_breaker):
        self.cache = cache
        self.circuit_breaker = circuit_breaker

    async def get_user_profile(self, user_id: str) -> Dict:
        cache_key = f"user_profile:{user_id}"

        # Try cache first with circuit breaker protection
        try:
            cached = await self.circuit_breaker.call(
                self.cache.get(cache_key)
            )
            if cached:
                return json.loads(cached)
        except CircuitBreakerOpen:
            # Cache is down, fallback to database
            pass

        # Fetch from database
        profile = await self.db.get_user_profile(user_id)

        # Cache for 1 hour (fire and forget)
        asyncio.create_task(
            self._safe_cache_set(cache_key, json.dumps(profile), 3600)
        )

        return profile

    async def _safe_cache_set(self, key: str, value: str, ttl: int):
        """Cache operation that doesn't affect main flow if it fails."""
        try:
            await self.cache.set(key, value, ttl)
        except Exception as e:
            logger.warning(f"Cache set failed: {e}")

# ❌ Fat Interface: Forces unnecessary dependencies
class OrderManagementInterface(ABC):
    # Customer operations
    @abstractmethod
    async def create_order(self, order_data: Dict) -> Order:
        pass

    @abstractmethod
    async def get_order_status(self, order_id: str) -> OrderStatus:
        pass

    # Admin operations
    @abstractmethod
    async def cancel_order(self, order_id: str, reason: str) -> bool:
        pass

    @abstractmethod
    async def process_refund(self, order_id: str, amount: float) -> bool:
        pass

    # Analytics operations
    @abstractmethod
    async def get_order_analytics(self, date_range: DateRange) -> Analytics:
        pass

    @abstractmethod
    async def export_orders(self, filters: Dict) -> FileStream:
        pass

    # Inventory operations
    @abstractmethod
    async def reserve_inventory(self, items: List[Item]) -> bool:
        pass

    @abstractmethod
    async def release_inventory(self, items: List[Item]) -> bool:
        pass

# ✅ Segregated Interfaces: Clients depend only on what they use
class CustomerOrderInterface(ABC):
    @abstractmethod
    async def create_order(self, order_data: Dict) -> Order:
        pass

    @abstractmethod
    async def get_order_status(self, order_id: str) -> OrderStatus:
        pass

    @abstractmethod
    async def get_order_history(self, customer_id: str) -> List[Order]:
        pass

class OrderAdministrationInterface(ABC):
    @abstractmethod
    async def cancel_order(self, order_id: str, reason: str) -> bool:
        pass

    @abstractmethod
    async def process_refund(self, order_id: str, amount: float) -> bool:
        pass

    @abstractmethod
    async def update_order_status(self, order_id: str, status: OrderStatus) -> bool:
        pass

class OrderAnalyticsInterface(ABC):
    @abstractmethod
    async def get_order_analytics(self, date_range: DateRange) -> Analytics:
        pass

    @abstractmethod
    async def export_orders(self, filters: Dict) -> FileStream:
        pass

class InventoryInterface(ABC):
    @abstractmethod
    async def reserve_inventory(self, items: List[Item]) -> bool:
        pass

    @abstractmethod
    async def release_inventory(self, items: List[Item]) -> bool:
        pass

    @abstractmethod
    async def check_availability(self, items: List[Item]) -> Dict[str, int]:
        pass

# Clients depend only on what they need
class MobileOrderService:
    def __init__(self, customer_orders: CustomerOrderInterface):
        self.customer_orders = customer_orders

    async def place_order(self, order_data: Dict) -> Order:
        return await self.customer_orders.create_order(order_data)

class AdminDashboardService:
    def __init__(self, 
                 admin_orders: OrderAdministrationInterface,
                 analytics: OrderAnalyticsInterface):
        self.admin_orders = admin_orders
        self.analytics = analytics

    async def generate_daily_report(self) -> Report:
        today = DateRange.today()
        analytics = await self.analytics.get_order_analytics(today)
        return Report.from_analytics(analytics)

# Implementation can support multiple interfaces
class OrderService(CustomerOrderInterface, 
                  OrderAdministrationInterface, 
                  OrderAnalyticsInterface):
    def __init__(self, repository: OrderRepository, 
                 inventory: InventoryInterface):
        self.repository = repository
        self.inventory = inventory

    # Implement all interface methods
    async def create_order(self, order_data: Dict) -> Order:
        # Reserve inventory first
        items = order_data['items']
        if not await self.inventory.reserve_inventory(items):
            raise InsufficientInventoryError()

        return await self.repository.create(order_data)

    # ... other implementations

# Abstractions
class EventPublisher(ABC):
    @abstractmethod
    async def publish(self, event: DomainEvent) -> bool:
        pass

class UserRepository(ABC):
    @abstractmethod
    async def save(self, user: User) -> User:
        pass

    @abstractmethod
    async def find_by_id(self, user_id: str) -> Optional[User]:
        pass

class NotificationService(ABC):
    @abstractmethod
    async def send_notification(self, notification: Notification) -> bool:
        pass

# High-level policy (business logic)
class UserRegistrationUseCase:
    def __init__(self, 
                 user_repository: UserRepository,
                 event_publisher: EventPublisher,
                 notification_service: NotificationService):
        self.user_repository = user_repository
        self.event_publisher = event_publisher
        self.notification_service = notification_service

    async def register_user(self, registration_data: Dict) -> User:
        # Business logic - stable, doesn't change with infrastructure
        user = User.from_registration_data(registration_data)

        # Validate business rules
        if await self._email_already_exists(user.email):
            raise EmailAlreadyExistsError()

        # Save user
        saved_user = await self.user_repository.save(user)

        # Publish domain event
        event = UserRegisteredEvent(
            user_id=saved_user.id,
            email=saved_user.email,
            timestamp=datetime.utcnow()
        )
        await self.event_publisher.publish(event)

        # Send welcome notification
        notification = WelcomeNotification(
            recipient=saved_user.email,
            user_name=saved_user.name
        )
        await self.notification_service.send_notification(notification)

        return saved_user

    async def _email_already_exists(self, email: str) -> bool:
        existing = await self.user_repository.find_by_email(email)
        return existing is not None

# Concrete implementations (details)
class PostgresUserRepository(UserRepository):
    def __init__(self, database_pool: asyncpg.Pool):
        self.db = database_pool

    async def save(self, user: User) -> User:
        query = """
        INSERT INTO users (id, email, name, created_at)
        VALUES ($1, $2, $3, $4)
        RETURNING *
        """
        async with self.db.acquire() as conn:
            row = await conn.fetchrow(
                query, user.id, user.email, user.name, user.created_at
            )
            return User.from_db_row(row)

    async def find_by_id(self, user_id: str) -> Optional[User]:
        query = "SELECT * FROM users WHERE id = $1"
        async with self.db.acquire() as conn:
            row = await conn.fetchrow(query, user_id)
            return User.from_db_row(row) if row else None

class KafkaEventPublisher(EventPublisher):
    def __init__(self, kafka_producer: AIOKafkaProducer):
        self.producer = kafka_producer

    async def publish(self, event: DomainEvent) -> bool:
        try:
            await self.producer.send(
                topic=event.topic,
                value=event.to_json().encode('utf-8'),
                key=event.aggregate_id.encode('utf-8')
            )
            return True
        except Exception as e:
            logger.error(f"Failed to publish event: {e}")
            return False

class EmailNotificationService(NotificationService):
    def __init__(self, email_client: EmailClient):
        self.email_client = email_client

    async def send_notification(self, notification: Notification) -> bool:
        return await self.email_client.send_email(
            to=notification.recipient,
            subject=notification.subject,
            body=notification.body
        )

# Dependency injection container
class Container:
    def __init__(self, config: Config):
        self.config = config
        self._instances = {}

    async def get_user_repository(self) -> UserRepository:
        if 'user_repository' not in self._instances:
            db_pool = await asyncpg.create_pool(self.config.database_url)
            self._instances['user_repository'] = PostgresUserRepository(db_pool)
        return self._instances['user_repository']

    async def get_event_publisher(self) -> EventPublisher:
        if 'event_publisher' not in self._instances:
            producer = AIOKafkaProducer(
                bootstrap_servers=self.config.kafka_brokers
            )
            await producer.start()
            self._instances['event_publisher'] = KafkaEventPublisher(producer)
        return self._instances['event_publisher']

    async def get_notification_service(self) -> NotificationService:
        if 'notification_service' not in self._instances:
            email_client = EmailClient(
                smtp_host=self.config.smtp_host,
                smtp_port=self.config.smtp_port,
                username=self.config.smtp_username,
                password=self.config.smtp_password
            )
            self._instances['notification_service'] = EmailNotificationService(email_client)
        return self._instances['notification_service']

    async def get_user_registration_use_case(self) -> UserRegistrationUseCase:
        return UserRegistrationUseCase(
            user_repository=await self.get_user_repository(),
            event_publisher=await self.get_event_publisher(),
            notification_service=await self.get_notification_service()
        )

# Security-enhanced implementation following SRP and DIP
class SecureAuthenticationService:
    def __init__(self, password_hasher: PasswordHasher, 
                 token_generator: TokenGenerator,
                 audit_logger: AuditLogger,
                 rate_limiter: RateLimiter):
        self.password_hasher = password_hasher
        self.token_generator = token_generator
        self.audit_logger = audit_logger
        self.rate_limiter = rate_limiter

    async def authenticate(self, email: str, password: str, 
                          client_ip: str) -> Optional[AuthToken]:
        # Rate limiting to prevent DoS attacks
        if not await self.rate_limiter.allow(f"auth:{client_ip}", limit=5, window=300):
            await self.audit_logger.log_security_event(
                event_type="RATE_LIMIT_EXCEEDED",
                client_ip=client_ip,
                email=email
            )
            raise RateLimitExceededError("Too many authentication attempts")

        # Input validation and sanitization
        if not self._validate_email_format(email):
            await self.audit_logger.log_security_event(
                event_type="INVALID_EMAIL_FORMAT",
                client_ip=client_ip,
                email=email
            )
            raise ValidationError("Invalid email format")

        # Fetch user with timing attack protection
        user = await self.user_repository.find_by_email(email)

        # Always hash password to prevent timing attacks
        password_valid = False
        if user:
            password_valid = await self.password_hasher.verify(
                password, user.password_hash
            )
        else:
            # Dummy hash to maintain consistent timing
            await self.password_hasher.hash("dummy_password_for_timing")

        if not user or not password_valid:
            await self.audit_logger.log_security_event(
                event_type="AUTHENTICATION_FAILED",
                client_ip=client_ip,
                email=email,
                reason="invalid_credentials"
            )
            return None

        # Check for account status
        if user.is_locked or user.requires_password_reset:
            await self.audit_logger.log_security_event(
                event_type="AUTHENTICATION_BLOCKED",
                client_ip=client_ip,
                email=email,
                reason=f"account_status_{user.status}"
            )
            return None

        # Generate secure token
        token = await self.token_generator.generate_token(
            user_id=user.id,
            scopes=user.permissions,
            expires_in=3600  # 1 hour
        )

        await self.audit_logger.log_security_event(
            event_type="AUTHENTICATION_SUCCESS",
            client_ip=client_ip,
            user_id=user.id,
            token_id=token.id
        )

        return token

    def _validate_email_format(self, email: str) -> bool:
        """Validate email format to prevent injection attacks."""
        import re
        pattern = r'^[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}$'
        return bool(re.match(pattern, email)) and len(email) <= 254

class AuthorizationService:
    def __init__(self, permission_cache: CacheInterface,
                 policy_engine: PolicyEngine):
        self.permission_cache = permission_cache
        self.policy_engine = policy_engine

    async def can_access_resource(self, user_id: str, resource: str, 
                                action: str, context: Dict = None) -> bool:
        """
        Implement fine-grained authorization with caching.
        Follows principle of least privilege.
        """
        # Check cache first for performance
        cache_key = f"authz:{user_id}:{resource}:{action}"
        cached_result = await self.permission_cache.get(cache_key)

        if cached_result is not None:
            return cached_result == "allowed"

        # Evaluate policies with context
        decision = await self.policy_engine.evaluate(
            user_id=user_id,
            resource=resource,
            action=action,
            context=context or {}
        )

        # Cache decision for 5 minutes
        await self.permission_cache.set(
            cache_key, 
            "allowed" if decision.allowed else "denied",
            ttl=300
        )

        return decision.allowed

# Secure data handling following SRP
class PersonalDataHandler:
    def __init__(self, encryption_service: EncryptionService,
                 data_classifier: DataClassifier):
        self.encryption_service = encryption_service
        self.data_classifier = data_classifier

    async def store_user_data(self, user_data: Dict) -> str:
        """Store user data with appropriate protection based on sensitivity."""
        classified_data = self.data_classifier.classify(user_data)

        protected_data = {}
        for field, value in classified_data.items():
            if value.sensitivity_level == SensitivityLevel.HIGH:
                # Encrypt PII fields
                protected_data[field] = await self.encryption_service.encrypt(
                    str(value.data)
                )
            elif value.sensitivity_level == SensitivityLevel.MEDIUM:
                # Hash identifiable but non-sensitive data
                protected_data[field] = self._hash_field(str(value.data))
            else:
                # Store low-sensitivity data as-is
                protected_data[field] = value.data

        return await self.repository.store(protected_data)

    def _hash_field(self, value: str) -> str:
        """One-way hash for semi-sensitive data."""
        import hashlib
        return hashlib.sha256(value.encode()).hexdigest()

# Performance-optimized implementation with metrics
import asyncio
import time
from typing import Dict, List, Optional
from dataclasses import dataclass
from enum import Enum

class CacheStrategy(Enum):
    READ_THROUGH = "read_through"
    WRITE_THROUGH = "write_through"
    WRITE_BEHIND = "write_behind"

@dataclass
class PerformanceMetrics:
    cache_hits: int = 0
    cache_misses: int = 0
    avg_response_time: float = 0.0
    error_rate: float = 0.0
    throughput: float = 0.0

class PerformanceOptimizedUserRepository(UserRepository):
    def __init__(self, 
                 base_repository: UserRepository, 
                 cache: CacheInterface,
                 metrics_collector: MetricsCollector,
                 strategy: CacheStrategy = CacheStrategy.READ_THROUGH):
        self.base_repository = base_repository
        self.cache = cache
        self.metrics = metrics_collector
        self.strategy = strategy
        self.connection_pool = ConnectionPool(max_connections=100)

    async def find_by_id(self, user_id: str) -> Optional[User]:
        start_time = time.perf_counter()
        cache_key = f"user:{user_id}"

        try:
            # Try cache first with timeout
            cached = await asyncio.wait_for(
                self.cache.get(cache_key), 
                timeout=0.1  # 100ms cache timeout
            )

            if cached:
                self.metrics.increment('cache.hits')
                response_time = time.perf_counter() - start_time
                self.metrics.histogram('response_time', response_time)
                return User.from_json(cached)

            self.metrics.increment('cache.misses')

            # Fallback to database with connection pooling
            async with self.connection_pool.acquire() as conn:
                user = await self.base_repository.find_by_id_with_connection(
                    user_id, conn
                )

            if user:
                # Async cache write (fire and forget)
                asyncio.create_task(
                    self._cache_user_async(cache_key, user)
                )

            response_time = time.perf_counter() - start_time
            self.metrics.histogram('response_time', response_time)
            return user

        except asyncio.TimeoutError:
            # Cache timeout, fallback to database
            self.metrics.increment('cache.timeouts')
            return await self.base_repository.find_by_id(user_id)
        except Exception as e:
            self.metrics.increment('errors')
            self.metrics.increment(f'error.{type(e).__name__}')
            response_time = time.perf_counter() - start_time
            self.metrics.histogram('error_response_time', response_time)
            raise

    async def _cache_user_async(self, cache_key: str, user: User):
        """Cache user data asynchronously to avoid blocking main request."""
        try:
            await self.cache.set(cache_key, user.to_json(), ttl=3600)
            self.metrics.increment('cache.async_writes.success')
        except Exception as e:
            self.metrics.increment('cache.async_writes.failed')
            # Log but don't propagate cache errors

class BatchingUserRepository(UserRepository):
    """Implements batch loading pattern for performance."""

    def __init__(self, base_repository: UserRepository):
        self.base_repository = base_repository
        self.batch_size = 50
        self.batch_timeout = 0.010  # 10ms
        self.pending_requests: Dict[str, List[asyncio.Future]] = {}
        self.batch_task = None

    async def find_by_id(self, user_id: str) -> Optional[User]:
        """Batch multiple requests together for database efficiency."""
        future = asyncio.Future()

        if user_id not in self.pending_requests:
            self.pending_requests[user_id] = []

        self.pending_requests[user_id].append(future)

        # Start batch processing if not already running
        if self.batch_task is None or self.batch_task.done():
            self.batch_task = asyncio.create_task(self._process_batch())

        return await future

    async def _process_batch(self):
        """Process batched requests efficiently."""
        await asyncio.sleep(self.batch_timeout)

        if not self.pending_requests:
            return

        user_ids = list(self.pending_requests.keys())[:self.batch_size]
        batch_requests = {}

        for user_id in user_ids:
            batch_requests[user_id] = self.pending_requests.pop(user_id)

        try:
            # Single database query for multiple users
            users = await self.base_repository.find_by_ids(user_ids)
            user_map = {user.id: user for user in users}

            # Resolve all futures
            for user_id, futures in batch_requests.items():
                user = user_map.get(user_id)
                for future in futures:
                    future.set_result(user)

        except Exception as e:
            # Propagate error to all pending requests
            for futures in batch_requests.values():
                for future in futures:
                    future.set_exception(e)

# Performance monitoring dashboard
class PerformanceDashboard:
    def __init__(self, metrics_collector: MetricsCollector):
        self.metrics = metrics_collector

    async def get_performance_summary(self) -> Dict:
        """Generate comprehensive performance report."""
        return {
            'cache_performance': {
                'hit_rate': await self._calculate_cache_hit_rate(),
                'avg_lookup_time': await self.metrics.get_avg('cache.lookup_time'),
                'error_rate': await self._calculate_cache_error_rate()
            },
            'database_performance': {
                'avg_query_time': await self.metrics.get_avg('db.query_time'),
                'connection_pool_usage': await self.metrics.get_current('db.pool.active'),
                'slow_queries': await self.metrics.get_count('db.slow_queries')
            },
            'api_performance': {
                'requests_per_second': await self.metrics.get_rate('api.requests'),
                'p95_response_time': await self.metrics.get_percentile('response_time', 95),
                'error_rate': await self._calculate_api_error_rate()
            }
        }

    async def _calculate_cache_hit_rate(self) -> float:
        hits = await self.metrics.get_count('cache.hits')
        misses = await self.metrics.get_count('cache.misses')
        total = hits + misses
        return (hits / total * 100) if total > 0 else 0.0

# Cost-optimized implementation using DIP and intelligent storage selection
from enum import Enum
from dataclasses import dataclass
from typing import Dict, Any
import asyncio

class StorageTier(Enum):
    HOT = "hot"           # Frequently accessed - Standard storage
    WARM = "warm"         # Occasionally accessed - IA storage  
    COLD = "cold"         # Rarely accessed - Glacier
    ARCHIVE = "archive"   # Long-term retention - Deep Archive

@dataclass
class StorageMetrics:
    access_frequency: int
    last_accessed: float
    size_bytes: int
    estimated_monthly_cost: float

class IntelligentStorageService(ABC):
    @abstractmethod
    async def store_file(self, file_data: bytes, metadata: Dict) -> str:
        pass

    @abstractmethod
    async def optimize_storage_costs(self) -> Dict[str, float]:
        pass

class AWSIntelligentStorageService(IntelligentStorageService):
    def __init__(self, s3_client, cost_analyzer: CostAnalyzer):
        self.s3_client = s3_client
        self.cost_analyzer = cost_analyzer

        # Cost per GB per month (USD) - as of 2024
        self.storage_costs = {
            StorageTier.HOT: 0.023,      # S3 Standard
            StorageTier.WARM: 0.0125,    # S3 Standard-IA
            StorageTier.COLD: 0.004,     # S3 Glacier Instant Retrieval
            StorageTier.ARCHIVE: 0.00099  # S3 Glacier Deep Archive
        }

    async def store_file(self, file_data: bytes, metadata: Dict) -> str:
        """Store file with optimal storage class based on access patterns."""

        # Analyze access patterns to determine optimal storage tier
        tier = await self._determine_optimal_tier(metadata)
        storage_class = self._map_tier_to_aws_class(tier)

        # Calculate expected monthly cost
        size_gb = len(file_data) / (1024**3)
        monthly_cost = size_gb * self.storage_costs[tier]

        # Add cost tracking metadata
        enriched_metadata = {
            **metadata,
            'storage_tier': tier.value,
            'estimated_monthly_cost': monthly_cost,
            'optimization_timestamp': time.time()
        }

        try:
            response = await self.s3_client.put_object(
                Body=file_data,
                StorageClass=storage_class,
                Metadata=enriched_metadata,
                **metadata
            )

            # Track cost metrics
            await self.cost_analyzer.track_storage_cost(
                object_key=response['Key'],
                tier=tier,
                size_bytes=len(file_data),
                monthly_cost=monthly_cost
            )

            return response['Key']

        except Exception as e:
            await self.cost_analyzer.track_error('storage_failure', metadata)
            raise

    async def _determine_optimal_tier(self, metadata: Dict) -> StorageTier:
        """Intelligent tier selection based on access patterns and cost."""

        access_frequency = metadata.get('expected_access_frequency', 'unknown')
        retention_period = metadata.get('retention_days', 365)
        file_type = metadata.get('content_type', '')

        # Machine learning model predictions for access patterns
        ml_prediction = await self.cost_analyzer.predict_access_pattern(metadata)

        # Decision logic based on multiple factors
        if access_frequency == 'high' or ml_prediction['daily_access_probability'] > 0.1:
            return StorageTier.HOT
        elif access_frequency == 'medium' or ml_prediction['weekly_access_probability'] > 0.05:
            return StorageTier.WARM
        elif retention_period > 1095:  # 3+ years
            return StorageTier.ARCHIVE
        else:
            return StorageTier.COLD

    def _map_tier_to_aws_class(self, tier: StorageTier) -> str:
        mapping = {
            StorageTier.HOT: 'STANDARD',
            StorageTier.WARM: 'STANDARD_IA',
            StorageTier.COLD: 'GLACIER',
            StorageTier.ARCHIVE: 'DEEP_ARCHIVE'
        }
        return mapping[tier]

    async def optimize_storage_costs(self) -> Dict[str, float]:
        """Analyze and optimize existing storage for cost savings."""

        optimization_results = {
            'potential_savings': 0.0,
            'objects_optimized': 0,
            'recommendations': []
        }

        # Analyze all stored objects
        objects = await self.s3_client.list_objects_v2()

        for obj in objects.get('Contents', []):
            current_class = obj.get('StorageClass', 'STANDARD')
            metadata = await self.s3_client.head_object(Key=obj['Key'])

            # Get actual access patterns from CloudWatch
            access_pattern = await self.cost_analyzer.get_access_pattern(obj['Key'])

            # Determine optimal tier based on actual usage
            optimal_tier = await self._determine_optimal_tier_from_usage(access_pattern)
            optimal_class = self._map_tier_to_aws_class(optimal_tier)

            if current_class != optimal_class:
                current_cost = self._calculate_monthly_cost(obj['Size'], current_class)
                optimal_cost = self._calculate_monthly_cost(obj['Size'], optimal_class)
                potential_saving = current_cost - optimal_cost

                if potential_saving > 0:
                    optimization_results['potential_savings'] += potential_saving
                    optimization_results['objects_optimized'] += 1
                    optimization_results['recommendations'].append({
                        'object_key': obj['Key'],
                        'current_class': current_class,
                        'recommended_class': optimal_class,
                        'monthly_savings': potential_saving
                    })

        return optimization_results

class CostOptimizedComputeService:
    """Implements cost-effective compute patterns using SOLID principles."""

    def __init__(self, 
                 lambda_client,
                 ec2_client, 
                 cost_optimizer: CostOptimizer):
        self.lambda_client = lambda_client
        self.ec2_client = ec2_client
        self.cost_optimizer = cost_optimizer

    async def execute_workload(self, workload: Dict) -> str:
        """Choose optimal compute option based on workload characteristics."""

        # Analyze workload to determine best compute option
        compute_recommendation = await self.cost_optimizer.analyze_workload(workload)

        if compute_recommendation.option == 'lambda':
            return await self._execute_on_lambda(workload)
        elif compute_recommendation.option == 'spot':
            return await self._execute_on_spot_instance(workload)
        else:
            return await self._execute_on_regular_instance(workload)

    async def _execute_on_lambda(self, workload: Dict) -> str:
        """Cost-effective for short, stateless workloads."""

        # Optimize memory allocation for cost
        optimal_memory = await self.cost_optimizer.get_optimal_lambda_memory(
            workload['estimated_duration'],
            workload['memory_profile']
        )

        response = await self.lambda_client.invoke(
            FunctionName=workload['function_name'],
            Payload=json.dumps(workload['payload']),
            InvocationType='RequestResponse'
        )

        # Track actual vs estimated costs
        actual_duration = response['Duration']
        actual_memory = response['MemorySize']
        cost = self._calculate_lambda_cost(actual_duration, actual_memory)

        await self.cost_optimizer.track_execution_cost(
            workload_id=workload['id'],
            compute_type='lambda',
            actual_cost=cost,
            estimated_cost=workload['estimated_cost']
        )

        return response['RequestId']

# Cost monitoring and alerting
class CostMonitoringDashboard:
    def __init__(self, cost_analyzer: CostAnalyzer):
        self.cost_analyzer = cost_analyzer

    async def generate_cost_report(self) -> Dict:
        """Generate comprehensive cost analysis report."""

        return {
            'storage_optimization': {
                'current_monthly_cost': await self.cost_analyzer.get_storage_costs(),
                'potential_savings': await self.cost_analyzer.calculate_potential_storage_savings(),
                'optimization_recommendations': await self.cost_analyzer.get_storage_recommendations()
            },
            'compute_optimization': {
                'lambda_costs': await self.cost_analyzer.get_lambda_costs(),
                'ec2_costs': await self.cost_analyzer.get_ec2_costs(),
                'underutilized_resources': await self.cost_analyzer.find_underutilized_resources()
            },
            'cost_trends': {
                'monthly_growth': await self.cost_analyzer.get_cost_growth_rate(),
                'cost_per_user': await self.cost_analyzer.get_cost_per_user(),
                'efficiency_metrics': await self.cost_analyzer.get_efficiency_metrics()
            }
        }

# Before: Monolithic legacy class
class LegacyOrderProcessor:
    def process_order(self, order_data):
        # Validation (SRP violation)
        if not order_data.get('customer_id'):
            raise ValueError("Customer ID required")

        # Database operations (SRP violation)
        cursor = self.db.cursor()
        cursor.execute("INSERT INTO orders ...", order_data)
        order_id = cursor.lastrowid

        # Email sending (SRP violation)
        smtp = smtplib.SMTP('localhost')
        smtp.sendmail(
            'noreply@company.com',
            order_data['email'],
            f"Order {order_id} confirmed"
        )

        # Inventory management (SRP violation)
        for item in order_data['items']:
            cursor.execute(
                "UPDATE inventory SET quantity = quantity - ? WHERE id = ?",
                (item['quantity'], item['id'])
            )

        # Payment processing (SRP violation)
        stripe.Charge.create(
            amount=order_data['total'],
            currency='usd',
            source=order_data['card_token']
        )

        return order_id

# After: SOLID-compliant refactoring using Strangler Fig pattern¹⁹
class ModernOrderProcessor:
    def __init__(self, 
                 validator: OrderValidator,
                 repository: OrderRepository,
                 notification_service: NotificationService,
                 inventory_service: InventoryService,
                 payment_service: PaymentService):
        self.validator = validator
        self.repository = repository
        self.notification_service = notification_service
        self.inventory_service = inventory_service
        self.payment_service = payment_service

    async def process_order(self, order_data: Dict) -> str:
        # Each responsibility is delegated to focused components
        validated_order = await self.validator.validate(order_data)

        # Reserve inventory first
        reservation = await self.inventory_service.reserve_items(
            validated_order.items
        )

        try:
            # Process payment
            payment_result = await self.payment_service.charge(
                amount=validated_order.total,
                token=validated_order.payment_token
            )

            # Save order
            order = await self.repository.create(validated_order)

            # Confirm inventory reservation
            await self.inventory_service.confirm_reservation(reservation.id)

            # Send notification
            await self.notification_service.send_order_confirmation(order)

            return order.id

        except PaymentFailedException:
            # Release inventory on payment failure
            await self.inventory_service.release_reservation(reservation.id)
            raise

# Event-driven communication following SOLID principles at Uber scale
class RideEventHandler:
    def __init__(self, 
                 driver_service: DriverService,
                 passenger_service: PassengerService,
                 pricing_service: PricingService,
                 notification_service: NotificationService,
                 metrics_collector: MetricsCollector):
        self.driver_service = driver_service
        self.passenger_service = passenger_service
        self.pricing_service = pricing_service
        self.notification_service = notification_service
        self.metrics = metrics_collector

    async def handle_ride_requested(self, event: RideRequestedEvent):
        """Handle ride request with fault tolerance and performance tracking."""
        start_time = time.perf_counter()

        try:
            # Parallel processing of independent operations
            driver_matching_task = asyncio.create_task(
                self.driver_service.find_optimal_drivers(
                    location=event.pickup_location,
                    ride_type=event.ride_type
                )
            )

            pricing_task = asyncio.create_task(
                self.pricing_service.calculate_dynamic_pricing(
                    pickup=event.pickup_location,
                    dropoff=event.dropoff_location,
                    time=event.timestamp
                )
            )

            # Wait for both with timeout
            drivers, pricing = await asyncio.wait_for(
                asyncio.gather(driver_matching_task, pricing_task),
                timeout=5.0  # 5 second SLA
            )

            # Send notifications asynchronously (fire and forget)
            asyncio.create_task(
                self._notify_drivers(drivers, event, pricing)
            )

            processing_time = time.perf_counter() - start_time
            self.metrics.histogram('ride_request.processing_time', processing_time)
            self.metrics.increment('ride_request.success')

        except asyncio.TimeoutError:
            self.metrics.increment('ride_request.timeout')
            # Fallback to basic matching without dynamic pricing
            await self._handle_timeout_fallback(event)
        except Exception as e:
            self.metrics.increment('ride_request.error')
            self.metrics.increment(f'ride_request.error.{type(e).__name__}')
            # Dead letter queue for failed events
            await self._send_to_dlq(event, str(e))

    async def _notify_drivers(self, drivers: List[Driver], 
                            event: RideRequestedEvent, pricing: PricingInfo):
        """Notify multiple drivers with rate limiting and delivery guarantees."""

        notification_tasks = []
        for driver in drivers[:5]:  # Notify top 5 drivers
            task = asyncio.create_task(
                self.notification_service.send_ride_offer(
                    driver_id=driver.id,
                    ride_details=event,
                    pricing=pricing,
                    timeout=30  # Driver has 30 seconds to respond
                )
            )
            notification_tasks.append(task)

        # Wait for all notifications with individual error handling
        results = await asyncio.gather(*notification_tasks, return_exceptions=True)

        success_count = sum(1 for r in results if not isinstance(r, Exception))
        self.metrics.gauge('driver_notifications.success_rate', 
                          success_count / len(results))

# Serverless function following SRP
import json
from typing import Dict, Any

def lambda_handler(event: Dict[str, Any], context: Any) -> Dict[str, Any]:
    """Single responsibility: user registration validation"""
    try:
        # Dependency injection through environment variables
        validator = UserValidator()
        registration_data = json.loads(event['body'])

        result = validator.validate(registration_data)

        return {
            'statusCode': 200 if result.is_valid else 400,
            'body': json.dumps({
                'valid': result.is_valid,
                'errors': result.errors
            })
        }
    except Exception as e:
        return {
            'statusCode': 500,
            'body': json.dumps({'error': str(e)})
        }

class ModelInterface(ABC):
    @abstractmethod
    async def predict(self, input_data: Dict) -> PredictionResult:
        pass

    @abstractmethod
    def get_model_metadata(self) -> ModelMetadata:
        pass

class RecommendationService:
    def __init__(self, model: ModelInterface):
        self.model = model  # Can be any ML model implementation

    async def get_recommendations(self, user_id: str) -> List[Recommendation]:
        user_features = await self._extract_user_features(user_id)
        prediction = await self.model.predict(user_features)
        return self._convert_to_recommendations(prediction)