Compare commits

..

1 Commits

Author SHA1 Message Date
ي
ef7874dcdc Fail demo startup when required API routes are missing 2026-03-30 07:56:05 +05:30
2 changed files with 61 additions and 20 deletions

View File

@@ -48,13 +48,6 @@ load_dotenv(backend_dir / '.env') # backend/.env
load_dotenv(project_root / '.env') # root .env (fallback)
load_dotenv() # CWD .env (fallback)
PODCAST_ONLY_DEMO_MODE = os.getenv("PODCAST_ONLY_DEMO_MODE", "false").lower() in {
"1",
"true",
"yes",
"on",
}
# Set up clean logging for end users
from logging_config import setup_clean_logging
setup_clean_logging()
@@ -190,9 +183,7 @@ rate_limiter = RateLimiter(window_seconds=60, max_requests=200)
frontend_serving = FrontendServing(app)
router_manager = RouterManager(app)
onboarding_manager = None
if not PODCAST_ONLY_DEMO_MODE:
onboarding_manager = OnboardingManager(app)
onboarding_manager = OnboardingManager(app)
# Middleware Order (FastAPI executes in REVERSE order of registration - LIFO):
# Registration order: 1. Monitoring 2. Rate Limit 3. API Key Injection
@@ -264,14 +255,7 @@ async def router_status():
# Onboarding management endpoints
@app.get("/api/onboarding/status")
async def onboarding_status():
"""Get onboarding manager status (or demo-mode disabled state)."""
if PODCAST_ONLY_DEMO_MODE:
return {
"enabled": False,
"status": "disabled",
"message": "Onboarding is disabled for podcast-only demo mode.",
"demo_mode": "podcast_only",
}
"""Get onboarding manager status."""
return onboarding_manager.get_onboarding_status()
# Include routers using modular utilities
@@ -478,7 +462,7 @@ async def serve_frontend():
async def startup_event():
"""Initialize services on startup."""
try:
startup_report = run_startup_health_routine()
startup_report = run_startup_health_routine(app)
if startup_report.get("status") != "healthy":
logger.error(f"Startup readiness finished with failures: {startup_report.get('errors', [])}")

View File

@@ -3,6 +3,8 @@ from datetime import datetime, timezone
from pathlib import Path
from typing import Any, Dict, List, Optional
from fastapi import FastAPI
from fastapi.routing import APIRoute
from loguru import logger
from sqlalchemy import inspect, text
@@ -49,6 +51,60 @@ def _record_check(checks: List[Dict[str, Any]], name: str, ok: bool, detail: str
checks.append({"name": name, "ok": ok, "detail": detail})
def _is_demo_mode() -> bool:
app_env = os.getenv("APP_ENV", os.getenv("ENV", os.getenv("DEPLOY_ENV", ""))).strip().lower()
if app_env == "demo":
return True
return _env_true("ALWRITY_DEMO_MODE", default=False)
def _check_required_demo_routes(
app: Optional[FastAPI],
checks: List[Dict[str, Any]],
errors: List[str],
) -> None:
if not _is_demo_mode():
_record_check(
checks,
"demo_required_routes",
True,
"Skipped (not in demo mode). Set APP_ENV=demo or ALWRITY_DEMO_MODE=true to enforce.",
)
return
if app is None:
errors.append(
"Demo startup route check could not run because FastAPI app context was not provided to startup health routine."
)
_record_check(checks, "demo_required_routes_context", False, "missing app context")
return
required_routes = {
"/api/subscription/plans": "GET",
"/api/podcast/projects": "GET",
}
available_routes = {
(route.path, method)
for route in app.router.routes
if isinstance(route, APIRoute)
for method in route.methods
}
missing: List[str] = []
for path, method in required_routes.items():
if (path, method) in available_routes:
_record_check(checks, f"demo_route_{path}_{method}", True, "route registered")
else:
missing.append(f"{method} {path}")
_record_check(checks, f"demo_route_{path}_{method}", False, "route missing")
if missing:
errors.append(
"Demo mode startup check failed. Missing required API endpoints: "
f"{', '.join(missing)}. Ensure subscription and podcast routers are imported and included during app setup."
)
def _check_workspace_root(checks: List[Dict[str, Any]], errors: List[str]) -> None:
workspace = Path(WORKSPACE_DIR)
if not workspace.exists():
@@ -144,7 +200,7 @@ def _check_db_access(checks: List[Dict[str, Any]], errors: List[str], warnings:
return candidate_user
def run_startup_health_routine() -> Dict[str, Any]:
def run_startup_health_routine(app: Optional[FastAPI] = None) -> Dict[str, Any]:
checks: List[Dict[str, Any]] = []
errors: List[str] = []
warnings: List[str] = []
@@ -152,6 +208,7 @@ def run_startup_health_routine() -> Dict[str, Any]:
_check_workspace_root(checks, errors)
if not errors:
_check_db_access(checks, errors, warnings)
_check_required_demo_routes(app, checks, errors)
status = "healthy" if not errors else "failed"
report = {