Spaces:
Running
Running
File size: 5,026 Bytes
167c070 da3be01 167c070 da3be01 167c070 | 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 147 148 | """Bootstrap — dependency-ordered init at app startup.
Loads embedding model, pre-warms LLM, starts job-queue worker, installs signal handlers.
Fail-fast in production, warn-and-continue in development.
"""
import asyncio
import logging
import os
import time
from contextlib import asynccontextmanager
from fastapi import FastAPI
from .config import get_settings
from .database.session import dispose_engine, init_engine
from .job_queue import get_worker
from .observability import get_metrics, setup_observability
from .services.cache import get_cache
from .services.cloudinary import get_cloudinary
from .services.guardrails import get_guardrails
from .services.llm import get_llm
from .services.pii_detector import get_pii_detector
from .shutdown import get_shutdown_manager
from .vectorstore.Qdrant import get_vector_store
logger = logging.getLogger("rga_auditor.bootstrap")
StartupError = RuntimeError
async def bootstrap_app() -> dict:
"""Initialize all subsystems in dependency order. Returns a status dict."""
s = get_settings()
setup_observability()
t0 = time.time()
status: dict = {"steps": []}
def step(name: str, fn):
t = time.time()
try:
result = fn()
dt = time.time() - t
status["steps"].append({"name": name, "ok": True, "ms": int(dt * 1000)})
logger.info("bootstrap[%s] OK in %dms", name, int(dt * 1000))
return result
except Exception as e:
dt = time.time() - t
status["steps"].append({"name": name, "ok": False, "ms": int(dt * 1000), "error": str(e)})
if s.is_production:
logger.critical("bootstrap[%s] FAILED: %s", name, e)
raise StartupError(f"bootstrap step '{name}' failed: {e}") from e
logger.warning("bootstrap[%s] failed (dev mode, continuing): %s", name, e)
return None
t_db = time.time()
try:
await init_engine()
status["steps"].append({"name": "database", "ok": True, "ms": int((time.time() - t_db) * 1000)})
logger.info("bootstrap[database] OK in %dms", int((time.time() - t_db) * 1000))
except Exception as e:
status["steps"].append({"name": "database", "ok": False, "ms": int((time.time() - t_db) * 1000), "error": str(e)})
if s.is_production:
logger.critical("bootstrap[database] FAILED: %s", e)
raise StartupError(f"database init failed: {e}") from e
logger.warning("bootstrap[database] failed (dev mode, continuing): %s", e)
# 2. cache
step("cache", get_cache)
step("vector_store", get_vector_store)
# 4. guardrails + pii (best-effort)
step("guardrails", get_guardrails)
step("pii_detector", get_pii_detector)
step("cloudinary", get_cloudinary)
# 5. LLM pre-warm (optional)
if s.skip_llm_prewarm:
logger.info("bootstrap[llm] SKIPPED (SKIP_LLM_PREWARM=true)")
else:
def _prewarm():
llm = get_llm()
try:
# Synchronous prewarm: we can't await here; the LLM client's chat is async
return llm
except Exception as e:
if s.is_production:
raise
logger.warning("LLM prewarm failed: %s", e)
step("llm", _prewarm)
# actually prewarm async
try:
await get_llm().chat("ping", system="You are a test. Reply with 'pong'.", max_tokens=10)
logger.info("LLM prewarm OK")
except Exception as e:
if s.is_production:
raise StartupError(f"LLM prewarm failed: {e}")
logger.warning("LLM prewarm failed: %s", e)
# 6. job queue worker
await get_worker().start()
# 7. signal handlers
try:
loop = asyncio.get_running_loop()
get_shutdown_manager().install_signal_handlers(loop)
except Exception as e:
logger.debug("signal handlers skipped: %s", e)
# 8. metrics timestamp
get_metrics().startup_timestamp.set(time.time())
status["ready_at"] = time.time()
status["ready_in_ms"] = int((time.time() - t0) * 1000)
logger.info("Application READY in %dms", status["ready_in_ms"])
return status
async def shutdown_app() -> None:
"""Tear down in reverse order."""
logger.info("Application shutting down")
try:
mgr = get_shutdown_manager()
mgr.begin_shutdown()
await mgr.wait_drain()
except Exception as e:
logger.warning("drain: %s", e)
try:
await get_worker().stop()
except Exception as e:
logger.warning("worker stop: %s", e)
try:
await dispose_engine()
except Exception as e:
logger.warning("db dispose: %s", e)
@asynccontextmanager
async def lifespan(app: FastAPI):
await bootstrap_app()
yield
await shutdown_app()
def install_bootstrap(app: FastAPI) -> None:
"""Attach the bootstrap lifespan to an existing FastAPI app."""
app.router.lifespan_context = lifespan
|