3deaa5ef40
Complete Flask → FastAPI migration with: - FastAPI app with session auth, Azure AD SSO, rate limiting - SQLite-backed session store (survives restarts) - Bulk AI metadata generation with SSE progress - Admin panel (user management, audit log, AI usage) - Subpath deployment support (ROOT_PATH config) - Docker + deploy.sh for production deployment - Test suite (auth, upload, templates, imports, admin, sessions) Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
107 lines
3 KiB
Python
107 lines
3 KiB
Python
"""FastAPI dependency injection providers."""
|
|
|
|
import logging
|
|
from typing import Optional, Dict
|
|
from fastapi import Depends, Request, HTTPException, status
|
|
|
|
from .config import Settings, get_settings
|
|
from .session.store import SessionStore
|
|
from .services.auth_service import AuthService
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
# Singletons (initialized once via lifespan)
|
|
_database = None
|
|
_session_store = None
|
|
_auth_service = None
|
|
|
|
|
|
def init_dependencies(settings: Settings):
|
|
"""Initialize singleton dependencies. Called once from app lifespan."""
|
|
global _database, _session_store, _auth_service
|
|
|
|
from src.database import Database
|
|
|
|
_database = Database(db_path=settings.DB_PATH)
|
|
_session_store = SessionStore(db_path=settings.SESSION_DB_PATH)
|
|
_auth_service = AuthService(database=_database)
|
|
|
|
logger.info("Dependencies initialized")
|
|
|
|
|
|
def get_database():
|
|
"""Get Database instance."""
|
|
if _database is None:
|
|
raise RuntimeError("Database not initialized")
|
|
return _database
|
|
|
|
|
|
def get_session_store() -> SessionStore:
|
|
"""Get SessionStore instance."""
|
|
if _session_store is None:
|
|
raise RuntimeError("SessionStore not initialized")
|
|
return _session_store
|
|
|
|
|
|
def get_auth_service() -> AuthService:
|
|
"""Get AuthService instance."""
|
|
if _auth_service is None:
|
|
raise RuntimeError("AuthService not initialized")
|
|
return _auth_service
|
|
|
|
|
|
async def get_current_user(request: Request) -> Dict:
|
|
"""FastAPI dependency: require authenticated user.
|
|
|
|
Replaces Flask's @login_required decorator.
|
|
Checks session cookie against database, returns user dict or raises 401.
|
|
"""
|
|
session_id = request.session.get("session_id")
|
|
if not session_id:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_401_UNAUTHORIZED,
|
|
detail="Not authenticated",
|
|
)
|
|
|
|
auth = get_auth_service()
|
|
db_session = auth.validate_session(session_id)
|
|
if not db_session:
|
|
# Session expired or invalid — clear it
|
|
request.session.clear()
|
|
raise HTTPException(
|
|
status_code=status.HTTP_401_UNAUTHORIZED,
|
|
detail="Session expired",
|
|
)
|
|
|
|
user_id = db_session["user_id"]
|
|
user = auth.get_user_by_id(user_id)
|
|
if not user:
|
|
request.session.clear()
|
|
raise HTTPException(
|
|
status_code=status.HTTP_401_UNAUTHORIZED,
|
|
detail="User not found",
|
|
)
|
|
|
|
return user
|
|
|
|
|
|
async def get_current_user_optional(request: Request) -> Optional[Dict]:
|
|
"""Same as get_current_user but returns None instead of raising."""
|
|
try:
|
|
return await get_current_user(request)
|
|
except HTTPException:
|
|
return None
|
|
|
|
|
|
async def get_current_admin(request: Request) -> Dict:
|
|
"""FastAPI dependency: require authenticated admin user.
|
|
|
|
Raises 403 if user is not an admin.
|
|
"""
|
|
user = await get_current_user(request)
|
|
if user.get("role") != "admin":
|
|
raise HTTPException(
|
|
status_code=status.HTTP_403_FORBIDDEN,
|
|
detail="Admin access required",
|
|
)
|
|
return user
|