62 lines
2.0 KiB
Python
62 lines
2.0 KiB
Python
"""Web route dependencies."""
|
|
|
|
from fastapi import Request, HTTPException, Depends
|
|
from fastapi.responses import RedirectResponse
|
|
from sqlalchemy.orm import Session
|
|
from ..database import get_db
|
|
from ..models.user import User, UserRole
|
|
|
|
|
|
async def get_current_user_web(request: Request, db: Session = Depends(get_db)) -> User:
|
|
"""Get current user from session for web routes."""
|
|
user_id = request.session.get("user_id")
|
|
|
|
if not user_id:
|
|
raise HTTPException(status_code=303, headers={"Location": "/login"})
|
|
|
|
user = db.query(User).filter(User.id == user_id).first()
|
|
|
|
if not user or not user.is_active:
|
|
request.session.clear()
|
|
raise HTTPException(status_code=303, headers={"Location": "/login"})
|
|
|
|
return user
|
|
|
|
|
|
async def require_user_web(
|
|
current_user: User = Depends(get_current_user_web)
|
|
) -> User:
|
|
"""Require any authenticated user for web routes."""
|
|
return current_user
|
|
|
|
|
|
async def require_admin_web(
|
|
current_user: User = Depends(get_current_user_web)
|
|
) -> User:
|
|
"""Require admin role for web routes."""
|
|
if not current_user.is_admin:
|
|
raise HTTPException(status_code=403, detail="Admin-Rechte erforderlich")
|
|
return current_user
|
|
|
|
|
|
async def require_super_admin_web(
|
|
current_user: User = Depends(get_current_user_web)
|
|
) -> User:
|
|
"""Require super admin role for web routes."""
|
|
if current_user.role != UserRole.SUPER_ADMIN:
|
|
raise HTTPException(status_code=403, detail="Super-Admin-Rechte erforderlich")
|
|
return current_user
|
|
|
|
|
|
def get_flashed_messages(request: Request) -> list[dict]:
|
|
"""Get and clear flash messages from session."""
|
|
messages = request.session.pop("flash_messages", [])
|
|
return messages
|
|
|
|
|
|
def flash(request: Request, message: str, category: str = "info"):
|
|
"""Add a flash message to session."""
|
|
if "flash_messages" not in request.session:
|
|
request.session["flash_messages"] = []
|
|
request.session["flash_messages"].append({"category": category, "message": message})
|