teoat's picture
Upload core/documentation/api_docs.py with huggingface_hub
2a6ebf8 verified
"""
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=[],
)