app-service/backend/app/crud/notification.py

"""CRUD operations for Notification model."""

import json
from datetime import datetime
from typing import Optional, List
from sqlalchemy.orm import Session
from sqlalchemy import func, desc

from app.models.notification import Notification
from app.schemas.notification import NotificationCreate


class CRUDNotification:
    """CRUD operations for Notification model."""

    def create(
        self,
        db: Session,
        *,
        obj_in: NotificationCreate
    ) -> Notification:
        """Create a new notification."""
        extra_data_str = json.dumps(obj_in.extra_data) if obj_in.extra_data else None

        db_obj = Notification(
            user_id=obj_in.user_id,
            title=obj_in.title,
            message=obj_in.message,
            type=obj_in.type,
            link=obj_in.link,
            extra_data=extra_data_str,
            is_read=False
        )
        db.add(db_obj)
        db.commit()
        db.refresh(db_obj)
        return db_obj

    def create_for_user(
        self,
        db: Session,
        *,
        user_id: str,
        title: str,
        message: Optional[str] = None,
        type: str = "info",
        link: Optional[str] = None,
        extra_data: Optional[dict] = None
    ) -> Notification:
        """Convenience method to create a notification for a user."""
        obj_in = NotificationCreate(
            user_id=user_id,
            title=title,
            message=message,
            type=type,
            link=link,
            extra_data=extra_data
        )
        return self.create(db, obj_in=obj_in)

    def create_for_all_users(
        self,
        db: Session,
        *,
        title: str,
        message: Optional[str] = None,
        type: str = "system",
        link: Optional[str] = None,
        extra_data: Optional[dict] = None
    ) -> int:
        """Create a notification for all users (system notification)."""
        from app.models.user import User

        users = db.query(User).filter(User.is_active == True).all()
        count = 0

        for user in users:
            self.create_for_user(
                db,
                user_id=user.id,
                title=title,
                message=message,
                type=type,
                link=link,
                extra_data=extra_data
            )
            count += 1

        return count

    def get(self, db: Session, id: str) -> Optional[Notification]:
        """Get a notification by ID."""
        return db.query(Notification).filter(Notification.id == id).first()

    def get_multi_by_user(
        self,
        db: Session,
        *,
        user_id: str,
        skip: int = 0,
        limit: int = 50,
        unread_only: bool = False
    ) -> List[Notification]:
        """Get notifications for a user."""
        query = db.query(Notification).filter(Notification.user_id == user_id)

        if unread_only:
            query = query.filter(Notification.is_read == False)

        return query.order_by(desc(Notification.created_at))\
            .offset(skip)\
            .limit(limit)\
            .all()

    def count_by_user(self, db: Session, user_id: str) -> int:
        """Count total notifications for a user."""
        return db.query(Notification).filter(Notification.user_id == user_id).count()

    def count_unread_by_user(self, db: Session, user_id: str) -> int:
        """Count unread notifications for a user."""
        return db.query(Notification)\
            .filter(Notification.user_id == user_id)\
            .filter(Notification.is_read == False)\
            .count()

    def mark_as_read(self, db: Session, *, id: str, user_id: str) -> Optional[Notification]:
        """Mark a notification as read."""
        db_obj = db.query(Notification)\
            .filter(Notification.id == id)\
            .filter(Notification.user_id == user_id)\
            .first()

        if db_obj and not db_obj.is_read:
            db_obj.is_read = True
            db_obj.read_at = datetime.utcnow()
            db.add(db_obj)
            db.commit()
            db.refresh(db_obj)

        return db_obj

    def mark_all_as_read(self, db: Session, *, user_id: str) -> int:
        """Mark all notifications as read for a user."""
        count = db.query(Notification)\
            .filter(Notification.user_id == user_id)\
            .filter(Notification.is_read == False)\
            .update({
                "is_read": True,
                "read_at": datetime.utcnow()
            })
        db.commit()
        return count

    def mark_multiple_as_read(
        self,
        db: Session,
        *,
        user_id: str,
        notification_ids: List[str]
    ) -> int:
        """Mark multiple notifications as read."""
        count = db.query(Notification)\
            .filter(Notification.id.in_(notification_ids))\
            .filter(Notification.user_id == user_id)\
            .filter(Notification.is_read == False)\
            .update({
                "is_read": True,
                "read_at": datetime.utcnow()
            }, synchronize_session=False)
        db.commit()
        return count

    def delete(self, db: Session, *, id: str, user_id: str) -> bool:
        """Delete a notification."""
        obj = db.query(Notification)\
            .filter(Notification.id == id)\
            .filter(Notification.user_id == user_id)\
            .first()

        if obj:
            db.delete(obj)
            db.commit()
            return True
        return False

    def delete_all_read(self, db: Session, *, user_id: str) -> int:
        """Delete all read notifications for a user."""
        count = db.query(Notification)\
            .filter(Notification.user_id == user_id)\
            .filter(Notification.is_read == True)\
            .delete()
        db.commit()
        return count

    def delete_multiple(
        self,
        db: Session,
        *,
        user_id: str,
        notification_ids: List[str]
    ) -> int:
        """Delete multiple notifications."""
        count = db.query(Notification)\
            .filter(Notification.id.in_(notification_ids))\
            .filter(Notification.user_id == user_id)\
            .delete(synchronize_session=False)
        db.commit()
        return count

    def get_stats_by_user(self, db: Session, user_id: str) -> dict:
        """Get notification statistics for a user."""
        total = self.count_by_user(db, user_id)
        unread = self.count_unread_by_user(db, user_id)

        # Count by type
        type_counts = db.query(
            Notification.type,
            func.count(Notification.id).label('count')
        ).filter(Notification.user_id == user_id)\
            .group_by(Notification.type)\
            .all()

        by_type = {t: c for t, c in type_counts}

        return {
            "total": total,
            "unread": unread,
            "by_type": by_type
        }


# Create instance
notification = CRUDNotification()