Spaces:
Running
Running
File size: 4,936 Bytes
e561008 b324c71 e561008 889e86b da1822e 1f5f31a b324c71 e561008 | 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 | import os
import json
from contextlib import asynccontextmanager
# ─── Load environment FIRST — before any local imports that read env vars ─────
# Routes instantiate Groq/Firebase clients at module level, so env vars must
# be present before those modules are imported.
from dotenv import load_dotenv
load_dotenv()
from fastapi import FastAPI, Request
from fastapi.middleware.cors import CORSMiddleware
from fastapi.responses import JSONResponse
import firebase_admin
from firebase_admin import credentials
from routes import bff, chat, transcribe, documents, consultation, planning, admin, account, help as help_routes, news, bsa as bsa_routes, bff_form
# ─── Firebase Admin SDK initialisation ───────────────────────────────────────
def _init_firebase():
if firebase_admin._apps:
return # already initialised (e.g. during hot-reload)
sa_json = os.environ.get("FIREBASE_SERVICE_ACCOUNT")
if sa_json:
try:
cred_dict = json.loads(sa_json)
cred = credentials.Certificate(cred_dict)
except json.JSONDecodeError as e:
raise RuntimeError(f"FIREBASE_SERVICE_ACCOUNT is not valid JSON: {e}")
else:
# Fall back to application default credentials (GCP environment)
cred = credentials.ApplicationDefault()
firebase_admin.initialize_app(cred)
_init_firebase()
# ─── App lifecycle ────────────────────────────────────────────────────────────
@asynccontextmanager
async def lifespan(app: FastAPI):
print("VenThrive API starting up...")
yield
print("VenThrive API shutting down.")
# ─── FastAPI app ──────────────────────────────────────────────────────────────
app = FastAPI(
title="VenThrive API",
version="1.0.0",
description="AI backend for VenThrive — Nigerian business operating system",
# Disable auto-generated docs in production
docs_url="/docs" if os.environ.get("ENV") != "production" else None,
redoc_url=None,
lifespan=lifespan,
)
# ─── CORS ─────────────────────────────────────────────────────────────────────
allowed_origins_str = os.environ.get("ALLOWED_ORIGINS", "http://localhost:3000")
allowed_origins = [o.strip() for o in allowed_origins_str.split(",")]
app.add_middleware(
CORSMiddleware,
allow_origins=allowed_origins,
allow_credentials=True,
allow_methods=["GET", "POST"],
allow_headers=["Authorization", "Content-Type"],
)
# ─── Global error handler ─────────────────────────────────────────────────────
@app.exception_handler(Exception)
async def global_exception_handler(request: Request, exc: Exception):
# Never expose internal error details to the client in production
return JSONResponse(
status_code=500,
content={"detail": "An unexpected error occurred. Please try again."},
)
# ─── Routes ───────────────────────────────────────────────────────────────────
app.include_router(bff.router, prefix="/api/bff", tags=["BFF"])
app.include_router(chat.router, prefix="/api/chat", tags=["Chat"])
app.include_router(transcribe.router, prefix="/api", tags=["Transcribe"])
app.include_router(documents.router, prefix="/api/documents", tags=["Documents"])
app.include_router(consultation.router, prefix="/api/consultation", tags=["Consultation"])
app.include_router(planning.router, prefix="/api/planning", tags=["Planning"])
app.include_router(admin.router, prefix="/api/admin", tags=["Admin"])
app.include_router(account.router, prefix="/api/account", tags=["Account"])
app.include_router(help_routes.router, prefix="/api/help", tags=["Help"])
app.include_router(news.router, prefix="/api/news", tags=["News"])
app.include_router(bsa_routes.router, prefix="/api/bsa", tags=["BSA"])
app.include_router(bff_form.router, prefix="/api/bff-form", tags=["BFF form"])
# ─── Health check (unauthenticated) ──────────────────────────────────────────
@app.get("/health", tags=["System"])
def health_check():
return {"status": "ok", "service": "venthrive-api"}
|