corpusdb / app /api_doc_generator.py
mrsavage1's picture
Upload 60 files
223a8c8 verified
from typing import List, Dict, Any
from datetime import datetime
class APIDocGenerator:
"""Auto-generate API documentation for databases"""
def generate_docs(self, database: str, tables: List[Dict], user_store) -> Dict[str, Any]:
"""Generate complete API documentation for a database"""
try:
from app.schema_manager import schema_manager
docs = {
'database': database,
'generated_at': datetime.utcnow().isoformat(),
'base_url': '/api',
'authentication': self._get_auth_docs(),
'endpoints': []
}
# Generate endpoints for each table
for table_info in tables:
if table_info.get('database') == database:
table_name = table_info['table']
schema = schema_manager.get(database, table_name, user_store)
table_docs = self._generate_table_endpoints(database, table_name, schema)
docs['endpoints'].extend(table_docs)
# Add database-level endpoints
docs['endpoints'].extend(self._get_database_endpoints(database))
return {'ok': True, 'docs': docs}
except Exception as e:
return {'ok': False, 'error': str(e)}
def _get_auth_docs(self) -> Dict[str, Any]:
"""Get authentication documentation"""
return {
'type': 'Bearer Token',
'description': 'Use session token or API key for authentication',
'methods': [
{
'name': 'Session Token',
'header': 'Authorization: Bearer <session_token>',
'example': 'Authorization: Bearer abc123xyz'
},
{
'name': 'API Key',
'headers': {
'X-Api-Key': '<api_key>',
'X-Secret-Key': '<secret_key>'
},
'example': {
'X-Api-Key': 'cDb_abc123',
'X-Secret-Key': 'secret-xyz-789'
}
}
]
}
def _generate_table_endpoints(self, database: str, table: str, schema: Dict) -> List[Dict]:
"""Generate CRUD endpoints for a table"""
columns = schema.get('columns', [])
column_names = [col['name'] for col in columns]
# Sample row for examples
sample_row = {col['name']: self._get_sample_value(col['type']) for col in columns}
endpoints = [
# List rows
{
'method': 'GET',
'path': f'/rows/{database}/{table}',
'description': f'Get rows from {table}',
'parameters': [
{'name': 'limit', 'type': 'integer', 'default': 100, 'description': 'Max rows to return'},
{'name': 'offset', 'type': 'integer', 'default': 0, 'description': 'Offset for pagination'}
],
'response': {
'type': 'array',
'example': [sample_row]
},
'curl_example': f'curl -H "Authorization: Bearer $TOKEN" {self._get_base_url()}/rows/{database}/{table}?limit=10'
},
# Create row
{
'method': 'POST',
'path': f'/rows/{database}/{table}',
'description': f'Insert a new row into {table}',
'body': {
'type': 'object',
'properties': {col['name']: {'type': col['type']} for col in columns},
'example': sample_row
},
'response': {
'type': 'object',
'example': {'ok': True, 'rows': 1}
},
'curl_example': f'''curl -X POST -H "Authorization: Bearer $TOKEN" \\
-H "Content-Type: application/json" \\
-d '{self._json_example(sample_row)}' \\
{self._get_base_url()}/rows/{database}/{table}'''
},
# Update row
{
'method': 'PUT',
'path': f'/rows/{database}/{table}/{{id_field}}/{{id_value}}',
'description': f'Update a row in {table}',
'parameters': [
{'name': 'id_field', 'type': 'string', 'description': 'Column name to identify row'},
{'name': 'id_value', 'type': 'string', 'description': 'Value to match'}
],
'body': {
'type': 'object',
'description': 'Fields to update',
'example': {column_names[0]: sample_row[column_names[0]]} if column_names else {}
},
'response': {
'type': 'object',
'example': {'ok': True, 'rows': 1}
},
'curl_example': f'''curl -X PUT -H "Authorization: Bearer $TOKEN" \\
-H "Content-Type: application/json" \\
-d '{{"name": "Updated Name"}}' \\
{self._get_base_url()}/rows/{database}/{table}/id/123'''
},
# Delete row
{
'method': 'DELETE',
'path': f'/rows/{database}/{table}/{{id_field}}/{{id_value}}',
'description': f'Delete a row from {table}',
'parameters': [
{'name': 'id_field', 'type': 'string', 'description': 'Column name to identify row'},
{'name': 'id_value', 'type': 'string', 'description': 'Value to match'}
],
'response': {
'type': 'object',
'example': {'ok': True, 'rows': 1}
},
'curl_example': f'curl -X DELETE -H "Authorization: Bearer $TOKEN" {self._get_base_url()}/rows/{database}/{table}/id/123'
}
]
return endpoints
def _get_database_endpoints(self, database: str) -> List[Dict]:
"""Get database-level endpoints"""
return [
{
'method': 'POST',
'path': '/sql',
'description': 'Execute custom SQL query',
'body': {
'type': 'object',
'properties': {
'sql': {'type': 'string', 'description': 'SQL query to execute'}
},
'example': {'sql': f'SELECT * FROM {database}.table_name LIMIT 10'}
},
'response': {
'type': 'object',
'example': {'ok': True, 'data': [], 'rows': 0}
},
'curl_example': f'''curl -X POST -H "Authorization: Bearer $TOKEN" \\
-H "Content-Type: application/json" \\
-d '{{"sql": "SELECT * FROM {database}.table_name LIMIT 10"}}' \\
{self._get_base_url()}/sql'''
}
]
def _get_sample_value(self, col_type: str) -> Any:
"""Get sample value for column type"""
type_upper = col_type.upper()
if 'INT' in type_upper:
return 123
elif 'FLOAT' in type_upper or 'DOUBLE' in type_upper or 'DECIMAL' in type_upper:
return 123.45
elif 'BOOL' in type_upper:
return True
elif 'DATE' in type_upper:
return '2024-01-15'
elif 'TIME' in type_upper:
return '14:30:00'
else:
return 'sample_value'
def _json_example(self, obj: Dict) -> str:
"""Format JSON example"""
import json
return json.dumps(obj, indent=2)
def _get_base_url(self) -> str:
"""Get base API URL"""
return 'https://your-space.hf.space/api'
def generate_sdk_examples(self, database: str, table: str) -> Dict[str, str]:
"""Generate SDK examples in multiple languages"""
return {
'python': f'''import requests
API_URL = "https://your-space.hf.space/api"
TOKEN = "your_token_here"
# List rows
response = requests.get(
f"{{API_URL}}/rows/{database}/{table}",
headers={{"Authorization": f"Bearer {{TOKEN}}"}}
)
data = response.json()
print(data)
# Insert row
new_row = {{"name": "John", "email": "john@example.com"}}
response = requests.post(
f"{{API_URL}}/rows/{database}/{table}",
headers={{"Authorization": f"Bearer {{TOKEN}}"}},
json=new_row
)
print(response.json())''',
'javascript': f'''const API_URL = "https://your-space.hf.space/api";
const TOKEN = "your_token_here";
// List rows
const response = await fetch(`${{API_URL}}/rows/{database}/{table}`, {{
headers: {{ Authorization: `Bearer ${{TOKEN}}` }}
}});
const data = await response.json();
console.log(data);
// Insert row
const newRow = {{ name: "John", email: "john@example.com" }};
const insertResponse = await fetch(`${{API_URL}}/rows/{database}/{table}`, {{
method: "POST",
headers: {{
Authorization: `Bearer ${{TOKEN}}`,
"Content-Type": "application/json"
}},
body: JSON.stringify(newRow)
}});
console.log(await insertResponse.json());''',
'php': f'''<?php
$apiUrl = "https://your-space.hf.space/api";
$token = "your_token_here";
// List rows
$ch = curl_init("$apiUrl/rows/{database}/{table}");
curl_setopt($ch, CURLOPT_HTTPHEADER, ["Authorization: Bearer $token"]);
curl_setopt($ch, CURLOPT_RETURNTRANSFER, true);
$response = curl_exec($ch);
$data = json_decode($response, true);
print_r($data);
// Insert row
$newRow = ["name" => "John", "email" => "john@example.com"];
$ch = curl_init("$apiUrl/rows/{database}/{table}");
curl_setopt($ch, CURLOPT_POST, true);
curl_setopt($ch, CURLOPT_HTTPHEADER, [
"Authorization: Bearer $token",
"Content-Type: application/json"
]);
curl_setopt($ch, CURLOPT_POSTFIELDS, json_encode($newRow));
curl_setopt($ch, CURLOPT_RETURNTRANSFER, true);
$response = curl_exec($ch);
print_r(json_decode($response, true));
?>''',
'curl': f'''# List rows
curl -H "Authorization: Bearer $TOKEN" \\
https://your-space.hf.space/api/rows/{database}/{table}
# Insert row
curl -X POST -H "Authorization: Bearer $TOKEN" \\
-H "Content-Type: application/json" \\
-d '{{"name": "John", "email": "john@example.com"}}' \\
https://your-space.hf.space/api/rows/{database}/{table}'''
}
api_doc_generator = APIDocGenerator()