feat: [Backend] Add most active user endpoint

web_app/api/serializers/leaderboard.py

@@ -0,0 +1,46 @@
+"""
+This module defines the API endpoints and serializers for the leaderboard functionality.
+"""
+from fastapi import APIRouter, Depends
+from pydantic import BaseModel
+from typing import List
+from sqlalchemy.orm import Session
+from web_app.db.crud.leaderboard import LeaderboardCRUD
+from web_app.db.session import get_db
+
+router = APIRouter()
+
+class UserLeaderboardItem(BaseModel):
+    """
+    Args:
+        db (Session): Database session dependency.
+
+    Returns:
+        UserLeaderboardResponse: Response containing the leaderboard data.
+    """
+    wallet_id: str
+    positions_number: int
+
+class UserLeaderboardResponse(BaseModel):
+    """
+    UserLeaderboardResponse is a model representing the response for a user leaderboard.
+
+    Attributes:
+        leaderboard (List[UserLeaderboardItem]): A list of user leaderboard items.
+    """
+    leaderboard: List[UserLeaderboardItem]
+
+@router.get(
+    "/api/get-user-leaderboard",
+    tags=["Leaderboard"],
+    response_model=UserLeaderboardResponse,
+    summary="Get user leaderboard",
+    response_description="Returns the top 10 users ordered by closed/opened positions.",
+)
+async def get_user_leaderboard(db: Session = Depends(get_db)) -> UserLeaderboardResponse:
+    """
+    Get the top 10 users ordered by closed/opened positions.
+    """
+    leaderboard_crud = LeaderboardCRUD(db)
+    leaderboard_data = leaderboard_crud.get_top_users_by_positions()
+    return UserLeaderboardResponse(leaderboard=leaderboard_data)

web_app/db/crud/leaderborad.py

@@ -0,0 +1,53 @@
+"""
+This module provides CRUD operations for the leaderboard, retrieving the top users by positions.
+
+"""
+from sqlalchemy.orm import Session
+from sqlalchemy.exc import SQLAlchemyError
+from sqlalchemy import func
+from web_app.db.models import User, Position
+import logging
+
+logger = logging.getLogger(__name__)
+
+class LeaderboardCRUD:
+    """
+    A class used to perform CRUD operations related to the leaderboard.
+    """
+    def __init__(self, session: Session):
+        """
+        Initializes a new instance of the class.
+
+        Args:
+            session (Session): The database session to be used for database operations.
+        """
+        self.Session = session
+
+    def get_top_users_by_positions(self) -> list[dict]:
+        """
+        Retrieves the top 10 users ordered by closed/opened positions.
+        :return: List of dictionaries containing wallet_id and positions_number.
+        """
+        with self.Session() as db:
+            try:
+                results = (
+                    db.query(
+                        User.wallet_id,
+                        func.count(Position.id).label("positions_number")
+                    )
+                    .join(Position, Position.user_id == User.id)
+                    .filter(Position.status.in_(["closed", "opened"]))
+                    .group_by(User.wallet_id)
+                    .order_by(func.count(Position.id).desc())
+                    .limit(10)
+                    .all()
+                )
+
+                return [
+                    {"wallet_id": result.wallet_id, "positions_number": result.positions_number}
+                    for result in results
+                ]
+
+            except SQLAlchemyError as e:
+                logger.error(f"Error retrieving top users by positions: {e}")
+                return []

web_app/db/crud/position.py

@@ -103,7 +103,7 @@ def get_positions_by_wallet_id(
             except SQLAlchemyError as e:
                 logger.error(f"Failed to retrieve positions: {str(e)}")
                 return []
-
+      
     def get_all_positions_by_wallet_id(
         self, wallet_id: str, start: int, limit: int
     ) -> list:
@@ -499,3 +499,5 @@ def get_extra_deposits_by_position_id(self, position_id: UUID) -> list[ExtraDepo
                 .all()
             )
             return extra_deposits
+
+