157 lines
4.7 KiB
Python
157 lines
4.7 KiB
Python
from fastapi import FastAPI
|
|
from fastapi.middleware.cors import CORSMiddleware
|
|
from fastapi.security import HTTPBearer
|
|
import logging
|
|
from sqlalchemy import text
|
|
from app.config.settings import settings
|
|
from app.config.database import engine
|
|
|
|
# Import models for Alembic
|
|
import app.models.auth.user_model
|
|
import app.models.auth.role_model
|
|
import app.models.auth.tenant_model
|
|
import app.models.theme.color_palette_model
|
|
|
|
# Configure logging
|
|
logging.basicConfig(
|
|
level=settings.LOG_LEVEL.upper(),
|
|
format="%(asctime)s - %(name)s - %(levelname)s - %(message)s",
|
|
datefmt="%Y-%m-%d %H:%M:%S",
|
|
)
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
def create_app() -> FastAPI:
|
|
app = FastAPI(
|
|
title=settings.PROJECT_NAME,
|
|
version=settings.VERSION,
|
|
description="SaaS Architecture API",
|
|
docs_url="/docs",
|
|
redoc_url="/redoc",
|
|
openapi_url="/openapi.json",
|
|
)
|
|
|
|
# === OpenAPI Security Scheme ===
|
|
from fastapi.openapi.utils import get_openapi
|
|
|
|
def custom_openapi():
|
|
if app.openapi_schema:
|
|
return app.openapi_schema
|
|
openapi_schema = get_openapi(
|
|
title=app.title,
|
|
version=app.version,
|
|
description=app.description,
|
|
routes=app.routes,
|
|
)
|
|
openapi_schema["components"]["securitySchemes"] = {
|
|
"BearerAuth": {
|
|
"type": "http",
|
|
"scheme": "bearer",
|
|
"bearerFormat": "JWT",
|
|
}
|
|
}
|
|
app.openapi_schema = openapi_schema
|
|
return app.openapi_schema
|
|
|
|
app.openapi = custom_openapi
|
|
|
|
# === CORS ===
|
|
origins = []
|
|
if settings.CORS_ALLOWED_ORIGINS:
|
|
origins = [
|
|
origin.strip()
|
|
for origin in settings.CORS_ALLOWED_ORIGINS.split(",")
|
|
if origin.strip()
|
|
]
|
|
|
|
if not origins:
|
|
raise RuntimeError(
|
|
"CORS_ALLOWED_ORIGINS must be set when allow_credentials=True"
|
|
)
|
|
|
|
app.add_middleware(
|
|
CORSMiddleware,
|
|
allow_origins=origins,
|
|
allow_credentials=True,
|
|
allow_methods=["*"],
|
|
allow_headers=["*"],
|
|
)
|
|
|
|
# === Include Routers ===
|
|
from app.routes.auth.auth import router as auth_router
|
|
from app.routes.auth.tenant import router as tenant_router
|
|
from app.routes.auth.role import router as role_router
|
|
from app.routes.auth.access import router as access_router
|
|
from app.routes.auth.user import router as user_router
|
|
|
|
app.include_router(auth_router, prefix="/api/auth", tags=["Authentication"])
|
|
app.include_router(tenant_router, prefix="/api/tenant", tags=["Tenant Management"])
|
|
app.include_router(role_router, prefix="/api/role", tags=["Role Management"])
|
|
app.include_router(access_router, prefix="/api/access", tags=["Access Management"])
|
|
app.include_router(user_router, prefix="/api/user", tags=["User Management"])
|
|
|
|
from app.routes.theme.color_palette import router as palette_router
|
|
app.include_router(palette_router, prefix="/api/theme", tags=["Theme Management"])
|
|
|
|
|
|
# === Startup: Test DB Connection (Sync + SQLAlchemy 2.0 compatible) ===
|
|
@app.on_event("startup")
|
|
def startup_event():
|
|
logger.info("Testing database connection...")
|
|
try:
|
|
with engine.connect() as conn:
|
|
conn.execute(text("SELECT 1"))
|
|
conn.commit()
|
|
logger.info("Database connection successful!")
|
|
except Exception as e:
|
|
logger.error(f"Database connection failed: {e}")
|
|
raise
|
|
|
|
logger.info(
|
|
f"{settings.PROJECT_NAME} v{settings.VERSION} started ({settings.APP_ENV})"
|
|
)
|
|
|
|
# === Basic Routes ===
|
|
@app.get("/", tags=["Root"])
|
|
def root():
|
|
"""
|
|
Root endpoint - API information.
|
|
|
|
Returns basic information about the API including version and documentation links.
|
|
"""
|
|
return {
|
|
"message": "Welcome to SaaS Architecture Backend API",
|
|
"version": settings.VERSION,
|
|
"docs": "/docs",
|
|
"redoc": "/redoc",
|
|
}
|
|
|
|
@app.get("/health", tags=["Health"])
|
|
def health():
|
|
"""
|
|
Health check endpoint.
|
|
|
|
Returns the health status of the API and database connection.
|
|
Used by monitoring tools and load balancers.
|
|
"""
|
|
db_status = "connected"
|
|
try:
|
|
with engine.connect() as conn:
|
|
conn.execute(text("SELECT 1"))
|
|
db_status = "healthy"
|
|
except Exception as e:
|
|
db_status = f"unhealthy: {str(e)}"
|
|
|
|
return {
|
|
"status": "healthy" if db_status == "healthy" else "degraded",
|
|
"environment": settings.APP_ENV,
|
|
"database": db_status,
|
|
"version": settings.VERSION,
|
|
}
|
|
|
|
return app
|
|
|
|
|
|
app = create_app()
|