.1915763867792649:1d1c1ff35df8ee6a3692d98ce6f396a6_69def9caf4ff2c686f6068ad.69defaf6f4ff2c686f6068fd.69defaf65d775e0eca8ed934:Trae CN.T(2026/4/15 10:41:58)

backend/app/alembic/versions/add_notification_table.py

@@ -0,0 +1,53 @@
+"""Add notification table
+
+Revision ID: add_notification_table
+Revises: 1a31ce608336
+Create Date: 2026-04-15 00:00:00.000000
+
+"""
+from alembic import op
+import sqlalchemy as sa
+import sqlmodel.sql.sqltypes
+from sqlalchemy.dialects import postgresql
+
+
+# revision identifiers, used by Alembic.
+revision = 'add_notification_table'
+down_revision = '1a31ce608336'
+branch_labels = None
+depends_on = None
+
+
+def upgrade():
+    # Create notification table
+    op.create_table(
+        'notification',
+        sa.Column('title', sqlmodel.sql.sqltypes.AutoString(length=255), nullable=False),
+        sa.Column('message', sqlmodel.sql.sqltypes.AutoString(length=1000), nullable=True),
+        sa.Column('notification_type', sa.Enum('INFO', 'SUCCESS', 'WARNING', 'ERROR', name='notificationtype'), nullable=False),
+        sa.Column('is_read', sa.Boolean(), nullable=False),
+        sa.Column('action_url', sqlmodel.sql.sqltypes.AutoString(length=500), nullable=True),
+        sa.Column('id', postgresql.UUID(as_uuid=True), nullable=False),
+        sa.Column('created_at', sa.DateTime(timezone=True), nullable=True),
+        sa.Column('updated_at', sa.DateTime(timezone=True), nullable=True),
+        sa.Column('user_id', postgresql.UUID(as_uuid=True), nullable=False),
+        sa.ForeignKeyConstraint(['user_id'], ['user.id'], ondelete='CASCADE'),
+        sa.PrimaryKeyConstraint('id')
+    )
+    # Create index on user_id for faster queries
+    op.create_index(op.f('ix_notification_user_id'), 'notification', ['user_id'], unique=False)
+    # Create index on created_at for ordering
+    op.create_index(op.f('ix_notification_created_at'), 'notification', ['created_at'], unique=False)
+    # Create index on is_read for filtering
+    op.create_index(op.f('ix_notification_is_read'), 'notification', ['is_read'], unique=False)
+
+
+def downgrade():
+    # Drop indexes first
+    op.drop_index(op.f('ix_notification_is_read'), table_name='notification')
+    op.drop_index(op.f('ix_notification_created_at'), table_name='notification')
+    op.drop_index(op.f('ix_notification_user_id'), table_name='notification')
+    # Drop table
+    op.drop_table('notification')
+    # Drop enum type
+    op.execute('DROP TYPE IF EXISTS notificationtype')

backend/app/api/main.py

@@ -1,13 +1,14 @@
 from fastapi import APIRouter
 
-from app.api.routes import items, login, private, users, utils
+from app.api.routes import items, login, notifications, private, users, utils
 from app.core.config import settings
 
 api_router = APIRouter()
 api_router.include_router(login.router)
 api_router.include_router(users.router)
 api_router.include_router(utils.router)
 api_router.include_router(items.router)
+api_router.include_router(notifications.router)
 
 
 if settings.ENVIRONMENT == "local":

backend/app/api/routes/notifications.py

@@ -0,0 +1,216 @@
+import uuid
+from typing import Any
+
+from fastapi import APIRouter, HTTPException
+from sqlmodel import col, func, select
+
+from app.api.deps import CurrentUser, SessionDep
+from app.core.websocket import manager
+from app.models import (
+    Message,
+    Notification,
+    NotificationCreate,
+    NotificationPublic,
+    NotificationsPublic,
+    NotificationUpdate,
+)
+
+router = APIRouter(prefix="/notifications", tags=["notifications"])
+
+
+@router.get("/", response_model=NotificationsPublic)
+def read_notifications(
+    session: SessionDep,
+    current_user: CurrentUser,
+    skip: int = 0,
+    limit: int = 100,
+    unread_only: bool = False,
+) -> Any:
+    """
+    Retrieve notifications for the current user.
+    """
+    if current_user.is_superuser:
+        count_statement = select(func.count()).select_from(Notification)
+        if unread_only:
+            count_statement = count_statement.where(Notification.is_read == False)
+        count = session.exec(count_statement).one()
+
+        unread_count_statement = (
+            select(func.count())
+            .select_from(Notification)
+            .where(Notification.is_read == False)
+        )
+        unread_count = session.exec(unread_count_statement).one()
+
+        statement = (
+            select(Notification)
+            .order_by(col(Notification.created_at).desc())
+            .offset(skip)
+            .limit(limit)
+        )
+        if unread_only:
+            statement = statement.where(Notification.is_read == False)
+        notifications = session.exec(statement).all()
+    else:
+        count_statement = (
+            select(func.count())
+            .select_from(Notification)
+            .where(Notification.user_id == current_user.id)
+        )
+        if unread_only:
+            count_statement = count_statement.where(Notification.is_read == False)
+        count = session.exec(count_statement).one()
+
+        unread_count_statement = (
+            select(func.count())
+            .select_from(Notification)
+            .where(Notification.user_id == current_user.id)
+            .where(Notification.is_read == False)
+        )
+        unread_count = session.exec(unread_count_statement).one()
+
+        statement = (
+            select(Notification)
+            .where(Notification.user_id == current_user.id)
+            .order_by(col(Notification.created_at).desc())
+            .offset(skip)
+            .limit(limit)
+        )
+        if unread_only:
+            statement = statement.where(Notification.is_read == False)
+        notifications = session.exec(statement).all()
+
+    notifications_public = [
+        NotificationPublic.model_validate(notification)
+        for notification in notifications
+    ]
+    return NotificationsPublic(
+        data=notifications_public, count=count, unread_count=unread_count
+    )
+
+
+@router.get("/{id}", response_model=NotificationPublic)
+def read_notification(
+    session: SessionDep, current_user: CurrentUser, id: uuid.UUID
+) -> Any:
+    """
+    Get notification by ID.
+    """
+    notification = session.get(Notification, id)
+    if not notification:
+        raise HTTPException(status_code=404, detail="Notification not found")
+    if not current_user.is_superuser and (notification.user_id != current_user.id):
+        raise HTTPException(status_code=403, detail="Not enough permissions")
+    return notification
+
+
+@router.post("/", response_model=NotificationPublic)
+async def create_notification(
+    *,
+    session: SessionDep,
+    current_user: CurrentUser,
+    notification_in: NotificationCreate,
+) -> Any:
+    """
+    Create new notification.
+    Only superusers can create notifications for other users.
+    Regular users can only create notifications for themselves.
+    """
+    if not current_user.is_superuser and notification_in.user_id != current_user.id:
+        raise HTTPException(
+            status_code=403,
+            detail="Not enough permissions to create notifications for other users",
+        )
+
+    notification = Notification.model_validate(notification_in)
+    session.add(notification)
+    session.commit()
+    session.refresh(notification)
+
+    notification_public = NotificationPublic.model_validate(notification)
+    await manager.send_notification(notification_public, notification.user_id)
+
+    return notification
+
+
+@router.put("/{id}", response_model=NotificationPublic)
+def update_notification(
+    *,
+    session: SessionDep,
+    current_user: CurrentUser,
+    id: uuid.UUID,
+    notification_in: NotificationUpdate,
+) -> Any:
+    """
+    Update a notification.
+    """
+    notification = session.get(Notification, id)
+    if not notification:
+        raise HTTPException(status_code=404, detail="Notification not found")
+    if not current_user.is_superuser and (notification.user_id != current_user.id):
+        raise HTTPException(status_code=403, detail="Not enough permissions")
+
+    update_dict = notification_in.model_dump(exclude_unset=True)
+    notification.sqlmodel_update(update_dict)
+    session.add(notification)
+    session.commit()
+    session.refresh(notification)
+    return notification
+
+
+@router.delete("/{id}")
+def delete_notification(
+    session: SessionDep, current_user: CurrentUser, id: uuid.UUID
+) -> Message:
+    """
+    Delete a notification.
+    """
+    notification = session.get(Notification, id)
+    if not notification:
+        raise HTTPException(status_code=404, detail="Notification not found")
+    if not current_user.is_superuser and (notification.user_id != current_user.id):
+        raise HTTPException(status_code=403, detail="Not enough permissions")
+    session.delete(notification)
+    session.commit()
+    return Message(message="Notification deleted successfully")
+
+
+@router.post("/{id}/mark-read", response_model=NotificationPublic)
+def mark_notification_read(
+    session: SessionDep, current_user: CurrentUser, id: uuid.UUID
+) -> Any:
+    """
+    Mark a notification as read.
+    """
+    notification = session.get(Notification, id)
+    if not notification:
+        raise HTTPException(status_code=404, detail="Notification not found")
+    if not current_user.is_superuser and (notification.user_id != current_user.id):
+        raise HTTPException(status_code=403, detail="Not enough permissions")
+
+    notification.is_read = True
+    session.add(notification)
+    session.commit()
+    session.refresh(notification)
+    return notification
+
+
+@router.post("/mark-all-read", response_model=Message)
+def mark_all_notifications_read(
+    session: SessionDep, current_user: CurrentUser
+) -> Message:
+    """
+    Mark all notifications as read for the current user.
+    """
+    statement = select(Notification).where(
+        Notification.user_id == current_user.id,
+        Notification.is_read == False,
+    )
+    notifications = session.exec(statement).all()
+
+    for notification in notifications:
+        notification.is_read = True
+        session.add(notification)
+
+    session.commit()
+    return Message(message=f"Marked {len(notifications)} notifications as read")

backend/app/api/routes/websocket.py

@@ -0,0 +1,81 @@
+import logging
+from typing import Annotated
+
+import jwt
+from fastapi import (
+    APIRouter,
+    HTTPException,
+    Query,
+    WebSocket,
+    WebSocketDisconnect,
+    status,
+)
+from jwt.exceptions import InvalidTokenError
+from pydantic import ValidationError
+from sqlmodel import Session
+
+from app.core import security
+from app.core.config import settings
+from app.core.db import engine
+from app.core.websocket import manager
+from app.models import TokenPayload, User
+
+logger = logging.getLogger(__name__)
+
+router = APIRouter(tags=["websocket"])
+
+
+def get_user_from_token(token: str) -> User:
+    try:
+        payload = jwt.decode(
+            token, settings.SECRET_KEY, algorithms=[security.ALGORITHM]
+        )
+        token_data = TokenPayload(**payload)
+    except (InvalidTokenError, ValidationError):
+        raise HTTPException(
+            status_code=status.HTTP_403_FORBIDDEN,
+            detail="Could not validate credentials",
+        )
+
+    with Session(engine) as session:
+        user = session.get(User, token_data.sub)
+        if not user:
+            raise HTTPException(status_code=404, detail="User not found")
+        if not user.is_active:
+            raise HTTPException(status_code=400, detail="Inactive user")
+        return user
+
+
+@router.websocket("/ws/notifications")
+async def websocket_notifications(
+    websocket: WebSocket,
+    token: Annotated[str | None, Query()] = None,
+) -> None:
+    """
+    WebSocket endpoint for real-time notifications.
+    Clients should connect with a valid JWT token as a query parameter.
+    """
+    if not token:
+        await websocket.close(code=status.WS_1008_POLICY_VIOLATION)
+        return
+
+    try:
+        user = get_user_from_token(token)
+    except HTTPException:
+        await websocket.close(code=status.WS_1008_POLICY_VIOLATION)
+        return
+
+    user_id = user.id
+    await manager.connect(user_id, websocket)
+
+    try:
+        while True:
+            data = await websocket.receive_text()
+            if data == "ping":
+                await websocket.send_text("pong")
+    except WebSocketDisconnect:
+        manager.disconnect(user_id, websocket)
+        logger.info(f"WebSocket disconnected for user: {user_id}")
+    except Exception as e:
+        manager.disconnect(user_id, websocket)
+        logger.error(f"WebSocket error for user {user_id}: {e}")