104 lines
3.8 KiB
Python
104 lines
3.8 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
Simple OpenAPI schema generator for FastAPI.
|
|
This generates the JSON schema file by fetching it from the running server's OpenAPI endpoint.
|
|
This ensures we get the complete schema as the server sees it at runtime.
|
|
"""
|
|
|
|
import json
|
|
import sys
|
|
import os
|
|
import asyncio
|
|
from pathlib import Path
|
|
import httpx
|
|
|
|
async def generate_schema_async():
|
|
"""Generate OpenAPI schema from the running FastAPI server"""
|
|
try:
|
|
# Configuration
|
|
public_url = os.getenv("PUBLIC_URL", "/")
|
|
if not public_url.endswith("/"):
|
|
public_url += "/"
|
|
|
|
# Server endpoint - use PUBLIC_SERVER_URL from .env
|
|
server_url = os.getenv("PUBLIC_SERVER_URL", "https://server:8000")
|
|
|
|
# Determine if SSL verification should be disabled based on protocol
|
|
use_ssl = server_url.startswith("https://")
|
|
verify_ssl = not use_ssl or os.getenv("SSL_VERIFY", "false").lower() == "true"
|
|
|
|
# OpenAPI schema endpoint
|
|
openapi_url = f"{server_url}{public_url}openapi.json"
|
|
|
|
print(f"🔍 Fetching OpenAPI schema from: {openapi_url}")
|
|
print(
|
|
f"🔒 SSL mode: {'enabled' if use_ssl else 'disabled'}, verification: {'enabled' if verify_ssl else 'disabled'}"
|
|
)
|
|
|
|
# Wait for server to be ready and fetch schema
|
|
max_retries = 10
|
|
retry_delay = 2
|
|
|
|
for attempt in range(max_retries):
|
|
try:
|
|
# Create HTTP client with appropriate SSL settings
|
|
async with httpx.AsyncClient(timeout=30.0, verify=verify_ssl) as client:
|
|
response = await client.get(openapi_url)
|
|
response.raise_for_status()
|
|
schema = response.json()
|
|
break
|
|
except (httpx.ConnectError, httpx.TimeoutException) as e:
|
|
if attempt < max_retries - 1:
|
|
print(
|
|
f"⏳ Server not ready (attempt {attempt + 1}/{max_retries}), waiting {retry_delay}s..."
|
|
)
|
|
await asyncio.sleep(retry_delay)
|
|
continue
|
|
else:
|
|
raise Exception(
|
|
f"Failed to connect to server after {max_retries} attempts: {e}"
|
|
)
|
|
except httpx.HTTPStatusError as e:
|
|
raise Exception(
|
|
f"Server returned error {e.response.status_code}: {e.response.text}"
|
|
)
|
|
else:
|
|
raise Exception("Failed to fetch schema - max retries exceeded")
|
|
|
|
# Write the schema to a JSON file that can be accessed from outside container
|
|
schema_file = Path("/client/openapi-schema.json")
|
|
with open(schema_file, 'w', encoding='utf-8') as f:
|
|
json.dump(schema, f, indent=2, ensure_ascii=False)
|
|
|
|
print(f"✅ OpenAPI schema generated successfully at: {schema_file}")
|
|
print(f"Schema contains {len(schema.get('paths', {}))} API paths")
|
|
print(f"Schema contains {len(schema.get('components', {}).get('schemas', {}))} component schemas")
|
|
|
|
# Print some info about what endpoints were found
|
|
paths = schema.get("paths", {})
|
|
print("📋 Found API endpoints:")
|
|
for path in sorted(paths.keys()):
|
|
methods = list(paths[path].keys())
|
|
print(f" {path} ({', '.join(methods)})")
|
|
|
|
return True
|
|
|
|
except Exception as e:
|
|
print(f"❌ Error generating OpenAPI schema: {e}")
|
|
import traceback
|
|
traceback.print_exc()
|
|
return False
|
|
|
|
def generate_schema():
|
|
"""Synchronous wrapper for async schema generation"""
|
|
try:
|
|
return asyncio.run(generate_schema_async())
|
|
except Exception as e:
|
|
print(f"❌ Error in schema generation wrapper: {e}")
|
|
return False
|
|
|
|
|
|
if __name__ == "__main__":
|
|
success = generate_schema()
|
|
sys.exit(0 if success else 1)
|