feat(tasks): support assignee filtering in team task listings

todo/constants/messages.py

@@ -81,6 +81,7 @@ class ValidationErrors:
     MISSING_EMAIL = "Email is required"
     MISSING_NAME = "Name is required"
     MISSING_PICTURE = "Picture is required"
+    TEAM_ID_REQUIRED_FOR_ASSIGNEE_FILTER = "teamId is required when filtering by assigneeId."
     SEARCH_QUERY_EMPTY = "Search query cannot be empty"
     TASK_ID_STRING_REQUIRED = "Task ID must be a string."
     INVALID_IS_ACTIVE_VALUE = "Invalid value for is_active"

todo/repositories/task_repository.py

@@ -29,6 +29,51 @@ def _get_team_task_ids(cls, team_id: str) -> List[ObjectId]:
         team_task_ids = [ObjectId(task["task_id"]) for task in team_tasks]
         return list(set(team_task_ids))
 
+    @classmethod
+    def _get_task_ids_for_assignees(cls, assignee_ids: List[str], team_id: str | None = None) -> List[ObjectId]:
+        """
+        Resolve active task IDs for the provided assignee IDs, optionally scoped to a team.
+        """
+        if not assignee_ids:
+            return []
+
+        candidate_values = set()
+        for assignee_id in assignee_ids:
+            candidate_values.add(assignee_id)
+            if ObjectId.is_valid(assignee_id):
+                candidate_values.add(ObjectId(assignee_id))
+
+        if not candidate_values:
+            return []
+
+        assignment_collection = TaskAssignmentRepository.get_collection()
+        assignment_filter: dict = {
+            "assignee_id": {"$in": list(candidate_values)},
+            "user_type": "user",
+            "is_active": True,
+        }
+
+        if team_id:
+            team_candidates = {team_id}
+            if ObjectId.is_valid(team_id):
+                team_candidates.add(ObjectId(team_id))
+            assignment_filter["team_id"] = {"$in": list(team_candidates)}
+
+        assignments = assignment_collection.find(
+            assignment_filter,
+            {"task_id": 1},
+        )
+
+        task_ids: set[ObjectId] = set()
+        for assignment in assignments:
+            task_identifier = assignment.get("task_id")
+            if isinstance(task_identifier, ObjectId):
+                task_ids.add(task_identifier)
+            elif isinstance(task_identifier, str) and ObjectId.is_valid(task_identifier):
+                task_ids.add(ObjectId(task_identifier))
+
+        return list(task_ids)
+
     @classmethod
     def _build_status_filter(cls, status_filter: str = None) -> dict:
         now = datetime.now(timezone.utc)
@@ -72,19 +117,41 @@ def list(
         user_id: str = None,
         team_id: str = None,
         status_filter: str = None,
+        assignee_ids: List[str] | None = None,
     ) -> List[TaskModel]:
         tasks_collection = cls.get_collection()
 
         base_filter = cls._build_status_filter(status_filter)
 
-        if team_id:
+        filters = [base_filter]
+
+        team_scope_applied = False
+
+        if assignee_ids:
+            assignee_task_ids = cls._get_task_ids_for_assignees(assignee_ids, team_id=team_id)
+            if not assignee_task_ids:
+                return []
+            filters.append({"_id": {"$in": assignee_task_ids}})
+            if team_id:
+                team_scope_applied = True
+        elif team_id:
             all_team_task_ids = cls._get_team_task_ids(team_id)
-            query_filter = {"$and": [base_filter, {"_id": {"$in": all_team_task_ids}}]}
-        elif user_id:
+            if not all_team_task_ids:
+                return []
+            filters.append({"_id": {"$in": all_team_task_ids}})
+            team_scope_applied = True
+
+        if user_id and not team_scope_applied:
             assigned_task_ids = cls._get_assigned_task_ids_for_user(user_id)
-            query_filter = {"$and": [base_filter, {"_id": {"$in": assigned_task_ids}}]}
+            user_filters = [{"createdBy": user_id}]
+            if assigned_task_ids:
+                user_filters.append({"_id": {"$in": assigned_task_ids}})
+            filters.append({"$or": user_filters})
+
+        if len(filters) == 1:
+            query_filter = filters[0]
         else:
-            query_filter = base_filter
+            query_filter = {"$and": filters}
 
         if sort_by == SORT_FIELD_UPDATED_AT:
             sort_direction = -1 if order == SORT_ORDER_DESC else 1
@@ -149,22 +216,47 @@ def _get_assigned_task_ids_for_user(cls, user_id: str) -> List[ObjectId]:
         return direct_task_ids + team_task_ids
 
     @classmethod
-    def count(cls, user_id: str = None, team_id: str = None, status_filter: str = None) -> int:
+    def count(
+        cls,
+        user_id: str = None,
+        team_id: str = None,
+        status_filter: str = None,
+        assignee_ids: List[str] | None = None,
+    ) -> int:
         tasks_collection = cls.get_collection()
 
         base_filter = cls._build_status_filter(status_filter)
 
-        if team_id:
+        filters = [base_filter]
+
+        team_scope_applied = False
+
+        if assignee_ids:
+            assignee_task_ids = cls._get_task_ids_for_assignees(assignee_ids, team_id=team_id)
+            if not assignee_task_ids:
+                return 0
+            filters.append({"_id": {"$in": assignee_task_ids}})
+            if team_id:
+                team_scope_applied = True
+        elif team_id:
             all_team_task_ids = cls._get_team_task_ids(team_id)
-            query_filter = {"$and": [base_filter, {"_id": {"$in": all_team_task_ids}}]}
+            if not all_team_task_ids:
+                return 0
+            filters.append({"_id": {"$in": all_team_task_ids}})
+            team_scope_applied = True
 
-        elif user_id:
+        if user_id and not team_scope_applied:
             assigned_task_ids = cls._get_assigned_task_ids_for_user(user_id)
-            query_filter = {
-                "$and": [base_filter, {"$or": [{"createdBy": user_id}, {"_id": {"$in": assigned_task_ids}}]}]
-            }
+            user_filters = [{"createdBy": user_id}]
+            if assigned_task_ids:
+                user_filters.append({"_id": {"$in": assigned_task_ids}})
+            filters.append({"$or": user_filters})
+
+        if len(filters) == 1:
+            query_filter = filters[0]
         else:
-            query_filter = base_filter
+            query_filter = {"$and": filters}
+
         return tasks_collection.count_documents(query_filter)
 
     @classmethod

todo/serializers/get_tasks_serializer.py

@@ -11,6 +11,19 @@ def to_internal_value(self, data):
         return super().to_internal_value(data)
 
 
+class QueryParameterListField(serializers.ListField):
+    """
+    DRF list field that understands QueryDict inputs with repeated parameters.
+    """
+
+    def get_value(self, dictionary):
+        if hasattr(dictionary, "getlist") and self.field_name in dictionary:
+            values = dictionary.getlist(self.field_name)
+            if values:
+                return values
+        return super().get_value(dictionary)
+
+
 class GetTaskQueryParamsSerializer(serializers.Serializer):
     page = serializers.IntegerField(
         required=False,
@@ -44,6 +57,11 @@ class GetTaskQueryParamsSerializer(serializers.Serializer):
 
     teamId = serializers.CharField(required=False, allow_blank=False, allow_null=True)
 
+    assigneeId = QueryParameterListField(
+        child=serializers.CharField(allow_blank=False),
+        required=False,
+    )
+
     status = CaseInsensitiveChoiceField(
         choices=[status.value for status in TaskStatus],
         required=False,
@@ -57,4 +75,8 @@ def validate(self, attrs):
             sort_by = validated_data.get("sort_by", SORT_FIELD_UPDATED_AT)
             validated_data["order"] = SORT_FIELD_DEFAULT_ORDERS[sort_by]
 
+        assignee_ids = validated_data.pop("assigneeId", None)
+        if assignee_ids is not None:
+            validated_data["assignee_ids"] = list(dict.fromkeys(assignee_ids))
+
         return validated_data

todo/services/task_service.py

@@ -74,6 +74,7 @@ def get_tasks(
         user_id: str,
         team_id: str = None,
         status_filter: str = None,
+        assignee_ids: List[str] | None = None,
     ) -> GetTasksResponse:
         try:
             cls._validate_pagination_params(page, limit)
@@ -93,9 +94,18 @@ def get_tasks(
                     )
 
             tasks = TaskRepository.list(
-                page, limit, sort_by, order, user_id, team_id=team_id, status_filter=status_filter
+                page,
+                limit,
+                sort_by,
+                order,
+                user_id,
+                team_id=team_id,
+                status_filter=status_filter,
+                assignee_ids=assignee_ids,
+            )
+            total_count = TaskRepository.count(
+                user_id, team_id=team_id, status_filter=status_filter, assignee_ids=assignee_ids
             )
-            total_count = TaskRepository.count(user_id, team_id=team_id, status_filter=status_filter)
 
             if not tasks:
                 return GetTasksResponse(tasks=[], links=None)

todo/views/task.py

@@ -28,6 +28,7 @@
 from todo.dto.responses.create_task_assignment_response import CreateTaskAssignmentResponse
 from todo.dto.task_assignment_dto import CreateTaskAssignmentDTO
 from todo.exceptions.task_exceptions import TaskNotFoundException
+from todo.repositories.team_repository import UserTeamDetailsRepository
 
 
 class TaskListView(APIView):
@@ -63,6 +64,14 @@ class TaskListView(APIView):
                 description="If provided, filters tasks by status (e.g., 'DONE', 'IN_PROGRESS', 'TODO', 'BLOCKED', 'DEFERRED').",
                 required=False,
             ),
+            OpenApiParameter(
+                name="assigneeId",
+                type=OpenApiTypes.STR,
+                location=OpenApiParameter.QUERY,
+                description="Repeatable parameter that filters tasks assigned to the provided user IDs.",
+                required=False,
+                many=True,
+            ),
         ],
         responses={
             200: OpenApiResponse(response=GetTasksResponse, description="Successful response"),
@@ -100,6 +109,22 @@ def get(self, request: Request):
 
         team_id = query.validated_data.get("teamId")
         status_filter = query.validated_data.get("status")
+        assignee_ids = query.validated_data.get("assignee_ids")
+
+        if assignee_ids:
+            if not team_id:
+                raise ValidationError(
+                    {"teamId": [ValidationErrors.TEAM_ID_REQUIRED_FOR_ASSIGNEE_FILTER]}
+                )
+
+            team_members = set(UserTeamDetailsRepository.get_users_by_team_id(team_id))
+            invalid_assignees = [assignee_id for assignee_id in assignee_ids if assignee_id not in team_members]
+
+            if invalid_assignees:
+                raise ValidationError(
+                    {"assigneeId": [f"{ValidationErrors.USER_NOT_TEAM_MEMBER}: {', '.join(invalid_assignees)}"]}
+                )
+
         response = TaskService.get_tasks(
             page=query.validated_data["page"],
             limit=query.validated_data["limit"],
@@ -108,6 +133,7 @@ def get(self, request: Request):
             user_id=request.user_id,
             team_id=team_id,
             status_filter=status_filter,
+            assignee_ids=assignee_ids,
         )
 
         if response.error and response.error.get("code") == "FORBIDDEN":