Devmain

app/api/endpoints/rating_endpoints.py

@@ -1,7 +1,10 @@
 from fastapi import APIRouter, Depends, HTTPException
 from sqlalchemy.orm import Session
 from app.api.auth import get_current_user
-from app.application.schemas.rating_dto import RatingCreateDTO, RatingUpdateDTO, RatingResponse
+from app.application.schemas.rating_dto import (
+    RatingCreateDTO, RatingUpdateDTO, RatingResponse,
+    RatingDistributionDTO, RecentRatingDTO, RatingStatsDTO
+)
 from app.application.schemas.user_dto import UserResponse
 from app.application.services.rating_service import RatingService
 from app.infrastructure.database import get_db
@@ -92,3 +95,24 @@ def delete_rating(rating_id: int, db: Session = Depends(get_db), token: str = De
     if not success:
         raise HTTPException(status_code=404, detail="Rating not found")
     return None
+
+# Analytics Endpoints
+@router.get("/distribution", response_model=list[RatingDistributionDTO])
+def get_rating_distribution(
+    db: Session = Depends(get_db),
+    role: str = Depends(require_role(["admin"]))):
+    rating_service = RatingService(db)
+    return rating_service.get_rating_distribution()
+
+@router.get("/recent", response_model=list[RecentRatingDTO])
+def get_recent_ratings(
+    limit: int = 10,
+    db: Session = Depends(get_db), role: str = Depends(require_role(["admin"]))):
+    rating_service = RatingService(db)
+    return rating_service.get_recent_ratings(limit)
+
+@router.get("/stats", response_model=RatingStatsDTO)
+def get_rating_stats(
+    db: Session = Depends(get_db), role: str = Depends(require_role(["admin"]))):
+    rating_service = RatingService(db)
+    return rating_service.get_rating_stats()

app/application/schemas/rating_dto.py

@@ -28,3 +28,33 @@ class RatingResponse(BaseModel):
     updated_at: datetime
 
     model_config = ConfigDict(from_attributes=True)
+
+# Analytics DTOs
+class RatingDistributionDTO(BaseModel):
+    value: int = Field(..., ge=1, le=5, description="Rating value (1-5)")
+    count: int = Field(..., description="Number of ratings with this value")
+
+class RecentRatingDTO(BaseModel):
+    id: int
+    value: int
+    item_name: str
+    user_name: str
+    created_at: datetime
+
+    class Config:
+        orm_mode = True
+
+class TopCategoryDTO(BaseModel):
+    name: str
+    count: int
+
+    class Config:
+        orm_mode = True
+
+class RatingStatsDTO(BaseModel):
+    average: float = Field(..., description="Average rating across all items")
+    totalCount: int = Field(..., description="Total number of ratings")
+    topCategory: TopCategoryDTO = Field(..., description="Category with most ratings")
+
+    class Config:
+        orm_mode = True

app/application/services/rating_service.py

@@ -1,10 +1,17 @@
 # app/application/services/rating_service.py
 
-from typing import List, Optional
+from typing import List, Optional, Dict, Any
 from sqlalchemy.orm import Session
+from sqlalchemy import func, desc
 from app.domain.rating import Rating
+from app.domain.item import Item
+from app.domain.user import User
+from app.domain.category import Category
 from app.infrastructure.repositories.rating_repository import RatingRepository
-from app.application.schemas.rating_dto import RatingCreateDTO, RatingUpdateDTO
+from app.application.schemas.rating_dto import (
+    RatingCreateDTO, RatingUpdateDTO, 
+    RatingDistributionDTO, RecentRatingDTO, RatingStatsDTO, TopCategoryDTO
+)
 
 class RatingService:
     def __init__(self, db_session: Session):
@@ -63,3 +70,15 @@ def remove_comment(self, rating_id: int):
         updated_rating = self.repository.update(rating_id, update_dto)
 
         return updated_rating
+
+    def get_rating_distribution(self) -> List[Dict[str, Any]]:
+        """Get distribution of ratings across values 1-5"""
+        return self.repository.get_rating_distribution()
+
+    def get_recent_ratings(self, limit: int = 10) -> List[Dict[str, Any]]:
+        """Get most recent ratings with user and item information"""
+        return self.repository.get_recent_ratings(limit)
+
+    def get_rating_stats(self) -> Dict[str, Any]:
+        """Get overall rating statistics"""
+        return self.repository.get_rating_stats()

app/infrastructure/repositories/rating_repository.py

@@ -1,10 +1,23 @@
-from typing import List, Optional
-from sqlalchemy.orm import Session
+from typing import List, Optional, Dict, Any
+from sqlalchemy import func, desc
+from sqlalchemy.orm import Session, joinedload
 from app.domain.rating import Rating
-from app.application.schemas.rating_dto import RatingCreateDTO, RatingUpdateDTO
+from app.domain.item import Item
+from app.domain.user import User
+from app.domain.category import Category
+from app.application.schemas.rating_dto import (
+    RatingCreateDTO, RatingUpdateDTO,
+    RatingDistributionDTO, TopCategoryDTO, RatingStatsDTO
+)
 
 class RatingRepository:
     def __init__(self, db: Session):
+        """
+        Initialize RatingRepository
+
+        Args:
+            db (Session): The SQLAlchemy session
+        """
         self.db = db
 
     def get_by_user_and_item(self, user_id: int, item_id: int) -> Rating | None:
@@ -53,3 +66,88 @@ def delete(self, rating_id: int) -> bool:
         self.db.delete(rating)
         self.db.commit()
         return True
+
+    def get_rating_distribution(self) -> List[RatingDistributionDTO]:
+        """
+        Get distribution of ratings across values 1-5
+        Returns a list with count for each rating value
+        """
+        results = self.db.query(
+            Rating.value,
+            func.count(Rating.id).label('count')
+        ).group_by(Rating.value).all()
+
+        # Make sure all values 1-5 are represented
+        distribution_map = {r.value: r.count for r in results}
+
+        distribution = []
+        for value in range(1, 6):
+            distribution.append(RatingDistributionDTO(
+                value=value,
+                count=distribution_map.get(value, 0)
+            ))
+
+        return distribution
+
+    def get_recent_ratings(self, limit: int = 10) -> List[Any]:
+        """
+        Get most recent ratings with user and item information
+        """
+        results = self.db.query(
+            Rating, Item.name.label('item_name'), User.name.label('user_name')
+        ).join(
+            Item, Rating.item_id == Item.id
+        ).join(
+            User, Rating.user_id == User.id
+        ).order_by(
+            desc(Rating.created_at)
+        ).limit(limit).all()
+
+        return [
+            {
+                'id': r.Rating.id,
+                'value': r.Rating.value,
+                'item_name': r.item_name,
+                'user_name': r.user_name,
+                'created_at': r.Rating.created_at
+            }
+            for r in results
+        ]
+
+    def get_rating_stats(self) -> Dict[str, Any]:
+        """
+        Get overall rating statistics
+        """
+        # Get average rating
+        avg_result = self.db.query(func.avg(Rating.value)).scalar() or 0
+        average = round(float(avg_result), 1)
+
+        # Get total count
+        total_count = self.db.query(func.count(Rating.id)).scalar() or 0
+
+        # Get top category - using the many-to-many relationship 
+        # between items and categories through the item_category table
+        top_category_result = self.db.query(
+            Category.name,
+            func.count(Rating.id).label('rating_count')
+        ).join(
+            # Use the proper association table to join categories and items
+            Item.categories
+        ).join(
+            Rating, Item.id == Rating.item_id
+        ).group_by(
+            Category.name
+        ).order_by(
+            desc('rating_count')
+        ).first()
+
+        top_category = {
+            'name': top_category_result.name if top_category_result else "Uncategorized",
+            'count': top_category_result.rating_count if top_category_result else 0
+        }
+
+        return RatingStatsDTO(
+            average=average,
+            totalCount=total_count,
+            topCategory=TopCategoryDTO(**top_category)
+        )