""" Enterprise API Versioning System for ETB-API Comprehensive versioning with backward compatibility and deprecation management """ import logging from datetime import datetime, timedelta from typing import Dict, List, Optional, Any, Union from django.http import JsonResponse, HttpRequest from django.conf import settings from django.utils import timezone from django.core.cache import cache from rest_framework import status from rest_framework.response import Response from rest_framework.views import APIView from rest_framework.decorators import api_view, permission_classes from rest_framework.permissions import AllowAny import json logger = logging.getLogger(__name__) class APIVersionManager: """Enterprise API version management system""" def __init__(self): self.supported_versions = getattr(settings, 'ALLOWED_VERSIONS', ['v1', 'v2']) self.default_version = getattr(settings, 'DEFAULT_VERSION', 'v1') self.deprecation_warning_days = 90 # 90 days warning before deprecation self.version_info = { 'v1': { 'status': 'stable', 'release_date': '2024-01-01', 'deprecation_date': None, 'sunset_date': None, 'features': [ 'incident_management', 'sla_monitoring', 'basic_analytics', 'user_management', 'basic_automation', ], 'endpoints': { 'incidents': '/api/v1/incidents/', 'sla': '/api/v1/sla/', 'analytics': '/api/v1/analytics/', 'users': '/api/v1/users/', 'automation': '/api/v1/automation/', }, 'changelog': [ { 'version': '1.0.0', 'date': '2024-01-01', 'changes': ['Initial release'], } ], }, 'v2': { 'status': 'beta', 'release_date': '2024-06-01', 'deprecation_date': None, 'sunset_date': None, 'features': [ 'incident_management', 'sla_monitoring', 'advanced_analytics', 'user_management', 'advanced_automation', 'ai_insights', 'real_time_collaboration', 'advanced_security', ], 'endpoints': { 'incidents': '/api/v2/incidents/', 'sla': '/api/v2/sla/', 'analytics': '/api/v2/analytics/', 'users': '/api/v2/users/', 'automation': '/api/v2/automation/', 'ai_insights': '/api/v2/ai-insights/', 'collaboration': '/api/v2/collaboration/', 'security': '/api/v2/security/', }, 'changelog': [ { 'version': '2.0.0-beta', 'date': '2024-06-01', 'changes': [ 'Added AI-powered incident insights', 'Enhanced real-time collaboration features', 'Improved security and compliance tools', 'Advanced analytics and reporting', ], } ], }, } def get_version_info(self, version: str) -> Optional[Dict[str, Any]]: """Get information about a specific API version""" return self.version_info.get(version) def get_supported_versions(self) -> List[str]: """Get list of supported API versions""" return self.supported_versions def is_version_supported(self, version: str) -> bool: """Check if a version is supported""" return version in self.supported_versions def is_version_deprecated(self, version: str) -> bool: """Check if a version is deprecated""" version_info = self.get_version_info(version) if not version_info: return False if version_info.get('deprecation_date'): deprecation_date = datetime.fromisoformat(version_info['deprecation_date']) return timezone.now() > deprecation_date return False def is_version_sunset(self, version: str) -> bool: """Check if a version is sunset (no longer available)""" version_info = self.get_version_info(version) if not version_info: return True # Unknown versions are considered sunset if version_info.get('sunset_date'): sunset_date = datetime.fromisoformat(version_info['sunset_date']) return timezone.now() > sunset_date return False def get_deprecation_warning(self, version: str) -> Optional[Dict[str, Any]]: """Get deprecation warning information for a version""" version_info = self.get_version_info(version) if not version_info: return None if version_info.get('deprecation_date'): deprecation_date = datetime.fromisoformat(version_info['deprecation_date']) days_until_deprecation = (deprecation_date - timezone.now()).days if days_until_deprecation <= self.deprecation_warning_days and days_until_deprecation > 0: return { 'warning': True, 'message': f'API version {version} will be deprecated in {days_until_deprecation} days', 'deprecation_date': version_info['deprecation_date'], 'recommended_version': self._get_recommended_version(version), } return None def _get_recommended_version(self, current_version: str) -> str: """Get recommended version for migration""" # Simple logic: recommend the latest stable version for version in reversed(self.supported_versions): version_info = self.get_version_info(version) if version_info and version_info.get('status') == 'stable': return version return self.default_version def get_migration_guide(self, from_version: str, to_version: str) -> Dict[str, Any]: """Get migration guide between versions""" from_info = self.get_version_info(from_version) to_info = self.get_version_info(to_version) if not from_info or not to_info: return { 'error': 'Invalid version specified', } # Generate migration guide based on version differences migration_guide = { 'from_version': from_version, 'to_version': to_version, 'breaking_changes': [], 'new_features': [], 'deprecated_features': [], 'migration_steps': [], } # Compare features from_features = set(from_info.get('features', [])) to_features = set(to_info.get('features', [])) migration_guide['new_features'] = list(to_features - from_features) migration_guide['deprecated_features'] = list(from_features - to_features) # Add specific migration steps based on version if from_version == 'v1' and to_version == 'v2': migration_guide['migration_steps'] = [ 'Update authentication headers to include API version', 'Replace deprecated endpoints with new v2 equivalents', 'Update request/response formats for new features', 'Implement new AI insights endpoints', 'Update collaboration features for real-time capabilities', ] migration_guide['breaking_changes'] = [ 'Authentication format changes', 'Some endpoint URLs have changed', 'Response format updates for analytics', ] return migration_guide def validate_request(self, request: HttpRequest) -> Dict[str, Any]: """Validate API request and return version information""" # Extract version from various sources version = self._extract_version_from_request(request) if not version: version = self.default_version # Check if version is supported if not self.is_version_supported(version): return { 'valid': False, 'error': f'Unsupported API version: {version}', 'supported_versions': self.supported_versions, } # Check if version is sunset if self.is_version_sunset(version): return { 'valid': False, 'error': f'API version {version} is no longer available', 'supported_versions': self.supported_versions, } # Get version info version_info = self.get_version_info(version) deprecation_warning = self.get_deprecation_warning(version) return { 'valid': True, 'version': version, 'version_info': version_info, 'deprecation_warning': deprecation_warning, } def _extract_version_from_request(self, request: HttpRequest) -> Optional[str]: """Extract API version from request""" # Check URL path path = request.path if '/api/v' in path: parts = path.split('/api/v') if len(parts) > 1: version_part = parts[1].split('/')[0] if version_part in self.supported_versions: return version_part # Check Accept header accept_header = request.META.get('HTTP_ACCEPT', '') if 'version=' in accept_header: for version in self.supported_versions: if f'version={version}' in accept_header: return version # Check custom header version_header = request.META.get('HTTP_X_API_VERSION') if version_header and version_header in self.supported_versions: return version_header # Check query parameter version_param = request.GET.get('version') if version_param and version_param in self.supported_versions: return version_param return None def add_deprecation_headers(self, response: Response, version: str) -> Response: """Add deprecation warning headers to response""" deprecation_warning = self.get_deprecation_warning(version) if deprecation_warning: response['X-API-Deprecation-Warning'] = deprecation_warning['message'] response['X-API-Deprecation-Date'] = deprecation_warning['deprecation_date'] response['X-API-Recommended-Version'] = deprecation_warning['recommended_version'] # Add version info headers version_info = self.get_version_info(version) if version_info: response['X-API-Version'] = version response['X-API-Status'] = version_info.get('status', 'unknown') response['X-API-Release-Date'] = version_info.get('release_date', '') return response # Global version manager instance version_manager = APIVersionManager() class VersionedAPIView(APIView): """Base class for versioned API views""" def dispatch(self, request, *args, **kwargs): """Override dispatch to handle versioning""" # Validate request version validation_result = version_manager.validate_request(request) if not validation_result['valid']: return Response( { 'error': validation_result['error'], 'supported_versions': validation_result.get('supported_versions', []), }, status=status.HTTP_400_BAD_REQUEST ) # Store version info in request request.api_version = validation_result['version'] request.version_info = validation_result['version_info'] request.deprecation_warning = validation_result.get('deprecation_warning') # Call parent dispatch response = super().dispatch(request, *args, **kwargs) # Add deprecation headers if hasattr(response, 'data'): response = version_manager.add_deprecation_headers(response, request.api_version) return response @api_view(['GET']) @permission_classes([AllowAny]) def api_version_info(request): """Get API version information""" version = request.GET.get('version') if version: # Get specific version info version_info = version_manager.get_version_info(version) if not version_info: return Response( {'error': f'Version {version} not found'}, status=status.HTTP_404_NOT_FOUND ) return Response({ 'version': version, 'info': version_info, 'deprecation_warning': version_manager.get_deprecation_warning(version), }) else: # Get all versions info all_versions = {} for version in version_manager.get_supported_versions(): all_versions[version] = { 'info': version_manager.get_version_info(version), 'deprecation_warning': version_manager.get_deprecation_warning(version), } return Response({ 'supported_versions': version_manager.get_supported_versions(), 'default_version': version_manager.default_version, 'versions': all_versions, }) @api_view(['GET']) @permission_classes([AllowAny]) def api_migration_guide(request): """Get migration guide between API versions""" from_version = request.GET.get('from') to_version = request.GET.get('to') if not from_version or not to_version: return Response( {'error': 'Both from and to parameters are required'}, status=status.HTTP_400_BAD_REQUEST ) migration_guide = version_manager.get_migration_guide(from_version, to_version) if 'error' in migration_guide: return Response( migration_guide, status=status.HTTP_400_BAD_REQUEST ) return Response(migration_guide) @api_view(['GET']) @permission_classes([AllowAny]) def api_changelog(request): """Get API changelog""" version = request.GET.get('version') if version: version_info = version_manager.get_version_info(version) if not version_info: return Response( {'error': f'Version {version} not found'}, status=status.HTTP_404_NOT_FOUND ) return Response({ 'version': version, 'changelog': version_info.get('changelog', []), }) else: # Get changelog for all versions all_changelogs = {} for version in version_manager.get_supported_versions(): version_info = version_manager.get_version_info(version) if version_info: all_changelogs[version] = version_info.get('changelog', []) return Response({ 'changelogs': all_changelogs, }) class APIVersionMiddleware: """Middleware to handle API versioning""" def __init__(self, get_response): self.get_response = get_response def __call__(self, request): # Only process API requests if request.path.startswith('/api/'): validation_result = version_manager.validate_request(request) if not validation_result['valid']: response = JsonResponse( { 'error': validation_result['error'], 'supported_versions': validation_result.get('supported_versions', []), }, status=400 ) return response # Store version info in request request.api_version = validation_result['version'] request.version_info = validation_result['version_info'] request.deprecation_warning = validation_result.get('deprecation_warning') response = self.get_response(request) # Add version headers to API responses if request.path.startswith('/api/') and hasattr(request, 'api_version'): response = version_manager.add_deprecation_headers(response, request.api_version) return response def version_required(versions: List[str]): """Decorator to require specific API versions""" def decorator(view_func): def wrapper(request, *args, **kwargs): if not hasattr(request, 'api_version'): return Response( {'error': 'API version not specified'}, status=status.HTTP_400_BAD_REQUEST ) if request.api_version not in versions: return Response( { 'error': f'API version {request.api_version} not supported for this endpoint', 'supported_versions': versions, }, status=status.HTTP_400_BAD_REQUEST ) return view_func(request, *args, **kwargs) return wrapper return decorator def deprecated_version(version: str, replacement_version: str): """Decorator to mark endpoints as deprecated in specific versions""" def decorator(view_func): def wrapper(request, *args, **kwargs): if hasattr(request, 'api_version') and request.api_version == version: # Add deprecation warning to response response = view_func(request, *args, **kwargs) if hasattr(response, 'data'): response['X-API-Deprecation-Warning'] = f'This endpoint is deprecated in {version}. Please use {replacement_version}.' response['X-API-Recommended-Version'] = replacement_version return response return view_func(request, *args, **kwargs) return wrapper return decorator class VersionedSerializer: """Base class for versioned serializers""" def __init__(self, version: str, *args, **kwargs): self.api_version = version super().__init__(*args, **kwargs) def get_versioned_fields(self, version: str) -> Dict[str, Any]: """Get fields specific to a version""" # Override in subclasses return {} def to_representation(self, instance): """Override to include version-specific fields""" data = super().to_representation(instance) # Add version-specific fields versioned_fields = self.get_versioned_fields(self.api_version) data.update(versioned_fields) return data class VersionedViewSet: """Mixin for versioned view sets""" def get_serializer_class(self): """Get serializer class based on API version""" if hasattr(self.request, 'api_version'): version = self.request.api_version versioned_serializer = getattr(self, f'serializer_class_{version}', None) if versioned_serializer: return versioned_serializer return super().get_serializer_class() def get_queryset(self): """Get queryset based on API version""" if hasattr(self.request, 'api_version'): version = self.request.api_version versioned_queryset = getattr(self, f'queryset_{version}', None) if versioned_queryset: return versioned_queryset return super().get_queryset() def list(self, request, *args, **kwargs): """Override list to add version-specific logic""" response = super().list(request, *args, **kwargs) # Add version-specific metadata if hasattr(request, 'api_version'): version_info = version_manager.get_version_info(request.api_version) if version_info: response.data['_meta'] = { 'api_version': request.api_version, 'version_status': version_info.get('status'), 'version_release_date': version_info.get('release_date'), } return response