557 lines
21 KiB
Python
557 lines
21 KiB
Python
"""
|
|
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
|