Spaces:
Paused
Paused
| """ | |
| Interactive API Documentation with OpenAPI 3.0 | |
| Enhanced developer experience with comprehensive docs | |
| """ | |
| from typing import Any, Dict, List | |
| from fastapi import FastAPI | |
| from fastapi.openapi.utils import get_openapi | |
| from fastapi.responses import HTMLResponse | |
| from core.logging.advanced_logging import structured_logger | |
| class EnhancedAPIDocumentation: | |
| """Enhanced API documentation with interactive features""" | |
| def __init__(self, app: FastAPI): | |
| self.app = app | |
| self.logger = structured_logger | |
| self.custom_examples = {} | |
| self.api_insights = {} | |
| def add_custom_examples(self, endpoint: str, examples: Dict[str, Any]): | |
| """Add custom examples for API endpoints""" | |
| self.custom_examples[endpoint] = examples | |
| def generate_enhanced_openapi(self) -> Dict[str, Any]: | |
| """Generate enhanced OpenAPI specification""" | |
| # Get base OpenAPI spec | |
| openapi_schema = get_openapi( | |
| title=self.app.title, | |
| version=self.app.version, | |
| description=self.app.description, | |
| routes=self.app.routes, | |
| ) | |
| # Enhance with custom metadata | |
| openapi_schema["info"]["contact"] = { | |
| "name": "Zenith Fraud Detection Platform", | |
| "email": "support@zenith-fraud.com", | |
| "url": "https://zenith-fraud.com" | |
| } | |
| openapi_schema["info"]["license"] = { | |
| "name": "Enterprise License", | |
| "url": "https://zenith-fraud.com/license" | |
| } | |
| openapi_schema["info"]["x-logo"] = { | |
| "url": "https://zenith-fraud.com/logo.png", | |
| "altText": "Zenith Fraud Detection Platform" | |
| } | |
| # Add security schemes | |
| openapi_schema["components"]["securitySchemes"] = { | |
| "bearerAuth": { | |
| "type": "http", | |
| "scheme": "bearer", | |
| "bearerFormat": "JWT", | |
| "description": "JWT Authorization header using the Bearer scheme" | |
| }, | |
| "apiKey": { | |
| "type": "apiKey", | |
| "in": "header", | |
| "name": "X-API-Key", | |
| "description": "API Key for service-to-service authentication" | |
| } | |
| } | |
| # Add global security | |
| openapi_schema["security"] = [ | |
| {"bearerAuth": []}, | |
| {"apiKey": []} | |
| ] | |
| # Enhance paths with examples and metadata | |
| if "paths" in openapi_schema: | |
| for path, methods in openapi_schema["paths"].items(): | |
| for method, operation in methods.items(): | |
| if method.upper() in ["GET", "POST", "PUT", "DELETE", "PATCH"]: | |
| # Add custom examples | |
| endpoint_key = f"{method.upper()} {path}" | |
| if endpoint_key in self.custom_examples: | |
| examples = self.custom_examples[endpoint_key] | |
| if "requestBody" not in operation and "parameters" in operation: | |
| # Add query parameter examples | |
| for param in operation.get("parameters", []): | |
| param_name = param.get("name") | |
| if param_name in examples: | |
| param["example"] = examples[param_name] | |
| # Add response examples | |
| if "responses" in operation: | |
| for status_code, response in operation["responses"].items(): | |
| if status_code == "200" and "content" in response: | |
| for content_type, content in response["content"].items(): | |
| if content_type == "application/json": | |
| # Add success response examples | |
| content["example"] = self._generate_response_example(operation, status_code) | |
| # Add operation metadata | |
| operation["x-code-samples"] = self._generate_code_samples(path, method.upper(), operation) | |
| return openapi_schema | |
| def _generate_response_example(self, operation: Dict[str, Any], status_code: str) -> Dict[str, Any]: | |
| """Generate response examples for documentation""" | |
| operation_id = operation.get("operationId", "") | |
| if "case" in operation_id.lower(): | |
| if "list" in operation_id.lower(): | |
| return { | |
| "cases": [ | |
| { | |
| "id": "case_12345", | |
| "title": "Suspicious Transaction Pattern", | |
| "status": "investigating", | |
| "priority": "high", | |
| "risk_score": 85.5, | |
| "created_at": "2024-01-15T10:30:00Z", | |
| "assignee": "investigator@example.com" | |
| } | |
| ], | |
| "total": 1, | |
| "page": 1, | |
| "page_size": 20 | |
| } | |
| elif "create" in operation_id.lower(): | |
| return { | |
| "id": "case_12345", | |
| "title": "New Suspicious Activity", | |
| "status": "open", | |
| "priority": "medium", | |
| "created_at": "2024-01-15T10:30:00Z", | |
| "message": "Case created successfully" | |
| } | |
| elif "auth" in operation_id.lower(): | |
| if "login" in operation_id.lower(): | |
| return { | |
| "access_token": "eyJ0eXAiOiJKV1QiLCJhbGciOiJIUzI1NiJ9...", | |
| "token_type": "bearer", | |
| "expires_in": 3600, | |
| "user": { | |
| "id": "user_123", | |
| "email": "user@example.com", | |
| "role": "investigator" | |
| } | |
| } | |
| # Default example | |
| return {"message": "Operation completed successfully"} | |
| def _generate_code_samples(self, path: str, method: str, operation: Dict[str, Any]) -> List[Dict[str, Any]]: | |
| """Generate code samples for different languages""" | |
| samples = [] | |
| # Python sample | |
| python_sample = f'''import requests | |
| url = "https://api.zenith-fraud.com{path}" | |
| headers = {{ | |
| "Authorization": "Bearer YOUR_JWT_TOKEN", | |
| "Content-Type": "application/json" | |
| }} | |
| response = requests.{method.lower()}(url, headers=headers) | |
| print(response.json())''' | |
| samples.append({ | |
| "lang": "python", | |
| "source": python_sample | |
| }) | |
| # JavaScript sample | |
| js_sample = f'''const response = await fetch('https://api.zenith-fraud.com{path}', {{ | |
| method: '{method}', | |
| headers: {{ | |
| 'Authorization': 'Bearer YOUR_JWT_TOKEN', | |
| 'Content-Type': 'application/json' | |
| }} | |
| }}); | |
| const data = await response.json(); | |
| console.log(data);''' | |
| samples.append({ | |
| "lang": "javascript", | |
| "source": js_sample | |
| }) | |
| # cURL sample | |
| curl_sample = f'''curl -X {method} \\ | |
| https://api.zenith-fraud.com{path} \\ | |
| -H "Authorization: Bearer YOUR_JWT_TOKEN" \\ | |
| -H "Content-Type: application/json"''' | |
| samples.append({ | |
| "lang": "curl", | |
| "source": curl_sample | |
| }) | |
| return samples | |
| class InteractiveDocumentation: | |
| """Interactive documentation with live API testing""" | |
| def __init__(self, app: FastAPI): | |
| self.app = app | |
| self.enhanced_docs = EnhancedAPIDocumentation(app) | |
| def get_enhanced_swagger_ui_html(self) -> HTMLResponse: | |
| """Get enhanced Swagger UI with custom features""" | |
| html_content = f""" | |
| <!DOCTYPE html> | |
| <html> | |
| <head> | |
| <title>Zenith Fraud Detection API - Interactive Documentation</title> | |
| <link rel="stylesheet" type="text/css" href="https://unpkg.com/swagger-ui-dist@5.7.2/swagger-ui.css" /> | |
| <style> | |
| .topbar {{ | |
| display: none !important; | |
| }} | |
| .info .title {{ | |
| color: #2c3e50 !important; | |
| }} | |
| .swagger-ui .info .description {{ | |
| margin-bottom: 20px; | |
| }} | |
| .api-explorer {{ | |
| background: #f8f9fa; | |
| padding: 20px; | |
| margin: 20px 0; | |
| border-radius: 8px; | |
| border-left: 4px solid #007bff; | |
| }} | |
| .performance-metrics {{ | |
| background: #e8f5e8; | |
| padding: 15px; | |
| margin: 15px 0; | |
| border-radius: 6px; | |
| border-left: 4px solid #28a745; | |
| }} | |
| .health-status {{ | |
| background: #fff3cd; | |
| padding: 10px; | |
| margin: 10px 0; | |
| border-radius: 4px; | |
| border-left: 4px solid #ffc107; | |
| }} | |
| </style> | |
| </head> | |
| <body> | |
| <div class="api-explorer"> | |
| <h2>🚀 Zenith Fraud Detection Platform API</h2> | |
| <p><strong>Version:</strong> {self.app.version}</p> | |
| <p><strong>Environment:</strong> <span id="environment-badge">Development</span></p> | |
| <div class="health-status"> | |
| <strong>API Health:</strong> <span id="health-status">Checking...</span> | |
| </div> | |
| <div class="performance-metrics"> | |
| <strong>Performance Metrics:</strong> | |
| <br>• Average Response Time: <span id="avg-response-time">Loading...</span> | |
| <br>• 95th Percentile: <span id="p95-response-time">Loading...</span> | |
| <br>• Success Rate: <span id="success-rate">Loading...</span> | |
| </div> | |
| </div> | |
| <div id="swagger-ui"></div> | |
| <script src="https://unpkg.com/swagger-ui-dist@5.7.2/swagger-ui-bundle.js"></script> | |
| <script> | |
| const ui = SwaggerUIBundle({{ | |
| url: '/openapi.json', | |
| dom_id: '#swagger-ui', | |
| deepLinking: true, | |
| presets: [ | |
| SwaggerUIBundle.presets.apis, | |
| SwaggerUIBundle.SwaggerUIStandalonePreset | |
| ], | |
| plugins: [ | |
| SwaggerUIBundle.plugins.DownloadUrl | |
| ], | |
| layout: "StandaloneLayout", | |
| validatorUrl: null, | |
| tryItOutEnabled: true, | |
| requestInterceptor: function(request) {{ | |
| // Add auth token if available | |
| const token = localStorage.getItem('api_token'); | |
| if (token) {{ | |
| request.headers.Authorization = 'Bearer ' + token; | |
| }} | |
| return request; | |
| }}, | |
| responseInterceptor: function(response) {{ | |
| // Store auth token from login responses | |
| if (response.url.includes('/auth/login') && response.status === 200) {{ | |
| try {{ | |
| const data = JSON.parse(response.data); | |
| if (data.access_token) {{ | |
| localStorage.setItem('api_token', data.access_token); | |
| }} | |
| }} catch (e) {{ | |
| // Ignore parsing errors | |
| }} | |
| }} | |
| return response; | |
| }} | |
| }}); | |
| // Update health status periodically | |
| async function updateHealthStatus() {{ | |
| try {{ | |
| const response = await fetch('/health/detailed'); | |
| const data = await response.json(); | |
| const statusElement = document.getElementById('health-status'); | |
| const overallStatus = data.status; | |
| statusElement.textContent = overallStatus.toUpperCase(); | |
| statusElement.style.color = overallStatus === 'healthy' ? '#28a745' : | |
| overallStatus === 'degraded' ? '#ffc107' : '#dc3545'; | |
| }} catch (e) {{ | |
| document.getElementById('health-status').textContent = 'ERROR'; | |
| document.getElementById('health-status').style.color = '#dc3545'; | |
| }} | |
| }} | |
| // Update performance metrics | |
| async function updatePerformanceMetrics() {{ | |
| try {{ | |
| const response = await fetch('/metrics'); | |
| const metrics = await response.text(); | |
| // Parse Prometheus metrics (simplified) | |
| const lines = metrics.split('\\n'); | |
| let avgResponseTime = 'N/A'; | |
| let p95ResponseTime = 'N/A'; | |
| let successRate = 'N/A'; | |
| lines.forEach(line => {{ | |
| if (line.includes('http_request_duration_seconds')) {{ | |
| // This would require more complex parsing in production | |
| }} | |
| }}); | |
| document.getElementById('avg-response-time').textContent = avgResponseTime; | |
| document.getElementById('p95-response-time').textContent = p95ResponseTime; | |
| document.getElementById('success-rate').textContent = successRate; | |
| }} catch (e) {{ | |
| console.log('Could not load performance metrics'); | |
| }} | |
| }} | |
| // Initialize | |
| updateHealthStatus(); | |
| updatePerformanceMetrics(); | |
| // Update every 30 seconds | |
| setInterval(updateHealthStatus, 30000); | |
| setInterval(updatePerformanceMetrics, 60000); | |
| </script> | |
| </body> | |
| </html> | |
| """ | |
| return HTMLResponse(html_content) | |
| def get_enhanced_redoc_html(self) -> HTMLResponse: | |
| """Get enhanced ReDoc documentation""" | |
| html_content = f""" | |
| <!DOCTYPE html> | |
| <html> | |
| <head> | |
| <title>Zenith Fraud Detection API - Reference Documentation</title> | |
| <meta charset="utf-8"/> | |
| <meta name="viewport" content="width=device-width, initial-scale=1"> | |
| <link href="https://fonts.googleapis.com/css?family=Montserrat:300,400,700|Roboto:300,400,700" rel="stylesheet"> | |
| <style> | |
| body {{ | |
| margin: 0; | |
| padding: 0; | |
| }} | |
| .api-header {{ | |
| background: linear-gradient(135deg, #667eea 0%, #764ba2 100%); | |
| color: white; | |
| padding: 40px 20px; | |
| text-align: center; | |
| }} | |
| .api-header h1 {{ | |
| margin: 0; | |
| font-size: 2.5em; | |
| font-weight: 300; | |
| }} | |
| .api-header p {{ | |
| margin: 10px 0 0 0; | |
| font-size: 1.2em; | |
| opacity: 0.9; | |
| }} | |
| .status-indicator {{ | |
| display: inline-block; | |
| padding: 5px 15px; | |
| border-radius: 20px; | |
| font-size: 0.9em; | |
| margin-left: 10px; | |
| }} | |
| .status-healthy {{ | |
| background: #28a745; | |
| color: white; | |
| }} | |
| .status-degraded {{ | |
| background: #ffc107; | |
| color: black; | |
| }} | |
| .status-unhealthy {{ | |
| background: #dc3545; | |
| color: white; | |
| }} | |
| </style> | |
| </head> | |
| <body> | |
| <div class="api-header"> | |
| <h1>Zenith Fraud Detection Platform</h1> | |
| <p>Enterprise API Reference Documentation</p> | |
| <div> | |
| <strong>Version {self.app.version}</strong> | |
| <span class="status-indicator status-healthy" id="api-status">HEALTHY</span> | |
| </div> | |
| </div> | |
| <redoc spec-url="/openapi.json"></redoc> | |
| <script src="https://cdn.redoc.ly/redoc/latest/bundles/redoc.standalone.js"></script> | |
| <script> | |
| // Initialize ReDoc | |
| Redoc.init('/openapi.json', {{ | |
| title: 'Zenith Fraud Detection API', | |
| theme: {{ | |
| colors: {{ | |
| primary: {{ | |
| main: '#667eea' | |
| }} | |
| }}, | |
| typography: {{ | |
| fontFamily: '"Roboto", "Helvetica", "Arial", sans-serif' | |
| }} | |
| }} | |
| }}, document.querySelector('redoc')); | |
| // Update API status | |
| async function updateApiStatus() {{ | |
| try {{ | |
| const response = await fetch('/health'); | |
| const data = await response.json(); | |
| const statusElement = document.getElementById('api-status'); | |
| const status = data.status.toUpperCase(); | |
| statusElement.textContent = status; | |
| statusElement.className = 'status-indicator status-' + data.status.toLowerCase(); | |
| }} catch (e) {{ | |
| document.getElementById('api-status').textContent = 'UNKNOWN'; | |
| document.getElementById('api-status').className = 'status-indicator status-unhealthy'; | |
| }} | |
| }} | |
| updateApiStatus(); | |
| setInterval(updateApiStatus, 30000); | |
| </script> | |
| </body> | |
| </html> | |
| """ | |
| return HTMLResponse(html_content) | |
| # Global documentation instance | |
| api_documentation = None | |
| def init_api_documentation(app: FastAPI) -> None: | |
| """Initialize enhanced API documentation""" | |
| global api_documentation | |
| api_documentation = InteractiveDocumentation(app) | |
| # Add custom examples | |
| api_documentation.enhanced_docs.add_custom_examples( | |
| "POST /api/v1/cases/", | |
| { | |
| "title": "Suspicious credit card transaction pattern", | |
| "description": "Multiple high-value transactions from unusual locations", | |
| "priority": "high", | |
| "risk_score": 88.5 | |
| } | |
| ) | |
| api_documentation.enhanced_docs.add_custom_examples( | |
| "GET /api/v1/cases/", | |
| { | |
| "status": "open", | |
| "priority": "high", | |
| "limit": 20, | |
| "offset": 0 | |
| } | |
| ) | |
| def get_enhanced_openapi_schema(): | |
| """Get enhanced OpenAPI schema""" | |
| if api_documentation: | |
| return api_documentation.enhanced_docs.generate_enhanced_openapi() | |
| return get_openapi( | |
| title="Zenith Fraud Detection API", | |
| version="1.0.0", | |
| description="Enterprise fraud detection and case management platform", | |
| routes=[], | |
| ) | |