File size: 4,632 Bytes
3d67472 e3892d4 ca1ff64 3d67472 e3892d4 3d67472 693f75f e3892d4 ca1ff64 e3892d4 693f75f e3892d4 693f75f 3d67472 693f75f 3d67472 693f75f ca1ff64 97a1a34 ca1ff64 693f75f e3892d4 693f75f 3d67472 ca1ff64 97a1a34 3d67472 e3892d4 |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 |
"""FastAPI main application module for AI Survey Simulator.
This module initializes the FastAPI application and configures:
- CORS middleware for frontend communication
- WebSocket endpoints for real-time conversation streaming
- REST API routes for conversation and persona management
- Database connections and session management
- Global exception handlers and logging
Typical usage:
uvicorn api.main:app --reload --host 0.0.0.0 --port 8000
"""
import logging
import sys
from pathlib import Path
from fastapi import FastAPI, WebSocket, Request
from fastapi.middleware.cors import CORSMiddleware
import uvicorn
# Ensure project root is available for shared config imports
ROOT_DIR = Path(__file__).resolve().parents[2]
if str(ROOT_DIR) not in sys.path:
sys.path.insert(0, str(ROOT_DIR))
from config.settings import get_settings # noqa: E402
# Import WebSocket endpoint and manager
from .conversation_ws import websocket_endpoint, manager # noqa: E402
from .routes import router as conversations_router # noqa: E402
from .conversation_service import initialize_conversation_service # noqa: E402
from backend.core.auth import COOKIE_NAME, INTERNAL_HEADER, get_app_password, verify_session_token # noqa: E402
# Load application settings
settings = get_settings()
# Setup logging using configured level
log_level = getattr(logging, settings.log_level.upper(), logging.INFO)
logging.basicConfig(level=log_level)
logger = logging.getLogger(__name__)
# Initialize FastAPI app
app = FastAPI(
title="AI Survey Simulator API",
description="Backend API for AI-to-AI healthcare survey conversations",
version="1.0.0"
)
# Configure CORS
app.add_middleware(
CORSMiddleware,
allow_origins=[
"http://localhost:7860", # Gradio default port
"http://127.0.0.1:7860", # Gradio alternative
],
allow_credentials=True,
allow_methods=["*"],
allow_headers=["*"],
)
# Include API routes
app.include_router(conversations_router)
@app.middleware("http")
async def auth_middleware(request: Request, call_next):
password = get_app_password()
if not password:
return await call_next(request)
path = request.url.path
if path in ("/health", "/docs", "/openapi.json"):
return await call_next(request)
if request.headers.get(INTERNAL_HEADER) == password:
return await call_next(request)
authz = request.headers.get("authorization")
if isinstance(authz, str) and authz.lower().startswith("bearer "):
token = authz.split(" ", 1)[1].strip()
if token and verify_session_token(token, password):
return await call_next(request)
token = request.cookies.get(COOKIE_NAME)
if token and verify_session_token(token, password):
return await call_next(request)
from fastapi.responses import JSONResponse
return JSONResponse({"error": "unauthorized"}, status_code=401)
@app.on_event("startup")
async def startup_event():
"""Initialize services on startup."""
logger.info("Initializing AI Survey Simulator API...")
# Initialize conversation service with WebSocket manager and settings
initialize_conversation_service(manager, settings)
logger.info("API startup complete")
@app.get("/")
async def root():
"""Root endpoint providing API information."""
return {
"message": "AI Survey Simulator API",
"version": "1.0.0",
"docs": "/docs",
"health": "/health"
}
@app.get("/health")
async def health_check():
"""Health check endpoint for monitoring."""
return {"status": "healthy"}
@app.websocket("/ws/conversation/{conversation_id}")
async def websocket_conversation_endpoint(websocket: WebSocket, conversation_id: str):
"""WebSocket endpoint for real-time conversation streaming.
Args:
websocket: WebSocket connection
conversation_id: Unique identifier for the conversation
"""
password = get_app_password()
if password:
if websocket.headers.get(INTERNAL_HEADER) == password:
pass
else:
bearer = websocket.query_params.get("token")
if isinstance(bearer, str) and bearer and verify_session_token(bearer, password):
pass
else:
token = websocket.cookies.get(COOKIE_NAME)
if not (token and verify_session_token(token, password)):
await websocket.close(code=1008)
return
await websocket_endpoint(websocket, conversation_id)
if __name__ == "__main__":
uvicorn.run(app, host=settings.api.host, port=settings.api.port)
|