77 lines
2.5 KiB
Python
77 lines
2.5 KiB
Python
from fastapi import APIRouter, Depends, HTTPException, Query
|
|
from typing import List, Optional, Dict, Any
|
|
import structlog
|
|
|
|
from app.schemas.notification import (
|
|
NotificationCreate,
|
|
NotificationResponse,
|
|
NotificationPreferences,
|
|
NotificationHistory
|
|
)
|
|
from app.services.notification_service import NotificationService
|
|
|
|
# Import unified authentication
|
|
from shared.auth.decorators import (
|
|
get_current_user_dep,
|
|
get_current_tenant_id_dep,
|
|
require_role
|
|
)
|
|
|
|
router = APIRouter(prefix="/notifications", tags=["notifications"])
|
|
logger = structlog.get_logger()
|
|
|
|
@router.post("/send", response_model=NotificationResponse)
|
|
async def send_notification(
|
|
notification: NotificationCreate,
|
|
tenant_id: str = Depends(get_current_tenant_id_dep),
|
|
current_user: Dict[str, Any] = Depends(get_current_user_dep),
|
|
):
|
|
"""Send notification to users"""
|
|
try:
|
|
logger.info("Sending notification",
|
|
tenant_id=tenant_id,
|
|
sender_id=current_user["user_id"],
|
|
type=notification.type)
|
|
|
|
notification_service = NotificationService()
|
|
|
|
# Ensure notification is scoped to tenant
|
|
notification.tenant_id = tenant_id
|
|
notification.sender_id = current_user["user_id"]
|
|
|
|
# Check permissions
|
|
if notification.broadcast and current_user.get("role") not in ["admin", "manager"]:
|
|
raise HTTPException(
|
|
status_code=403,
|
|
detail="Only admins and managers can send broadcast notifications"
|
|
)
|
|
|
|
result = await notification_service.send_notification(notification)
|
|
|
|
return result
|
|
|
|
except HTTPException:
|
|
raise
|
|
except Exception as e:
|
|
logger.error("Failed to send notification", error=str(e))
|
|
raise HTTPException(status_code=500, detail=str(e))
|
|
|
|
@router.get("/preferences", response_model=NotificationPreferences)
|
|
async def get_notification_preferences(
|
|
tenant_id: str = Depends(get_current_tenant_id_dep),
|
|
current_user: Dict[str, Any] = Depends(get_current_user_dep),
|
|
):
|
|
"""Get user's notification preferences"""
|
|
try:
|
|
notification_service = NotificationService()
|
|
|
|
preferences = await notification_service.get_user_preferences(
|
|
user_id=current_user["user_id"],
|
|
tenant_id=tenant_id
|
|
)
|
|
|
|
return preferences
|
|
|
|
except Exception as e:
|
|
logger.error("Failed to get preferences", error=str(e))
|
|
raise HTTPException(status_code=500, detail=str(e)) |