from typing import Callable from fastapi import Depends, HTTPException, Request from sqlalchemy.orm import Session from app.core.auth import decode_token, get_token_from_request from app.core.database import get_db from app.modules.auth.models import User from app.modules.auth.service import get_user class RequiresLoginException(Exception): pass def check_permission(user: User | None, permissions: list[str]) -> bool: if "public" in permissions: return True if user is None: return False if "authenticated" in permissions: return True if "admin" in permissions and user.is_admin: return True return False # group membership check: Part 2 async def get_current_user_optional( request: Request, db: Session = Depends(get_db) ) -> User | None: token = get_token_from_request(request) if not token: return None payload = decode_token(token) if not payload: return None user = get_user(db, payload.get("sub", "")) if user is None or not user.is_active: return None return user def require_permission(permissions: list[str]) -> Callable: async def _dep( request: Request, db: Session = Depends(get_db) ) -> User | None: user = await get_current_user_optional(request, db) if not check_permission(user, permissions): raise RequiresLoginException() return user return _dep async def get_current_user( request: Request, db: Session = Depends(get_db) ) -> User: token = get_token_from_request(request) if not token: raise RequiresLoginException() payload = decode_token(token) if not payload: raise RequiresLoginException() user = get_user(db, payload.get("sub", "")) if user is None or not user.is_active: raise RequiresLoginException() return user async def require_admin(user: User = Depends(get_current_user)) -> User: if not user.is_admin: raise HTTPException(status_code=403, detail="Admin access required") return user