feat: Get Subjects by Respondent Subject: Submissions + Assignments (M…

…2-7919) (#1611)

This PR creates the new endpoint `/subjects/respondent/{respondent_subject_id}/activity-or-flow/{activity_or_flow_id}`, which returns a list of subjects based on an association with the respondent subject and activity/flow specified. The endpoint is accessible by users with the following applet roles:
- Owner
- Manager
- Coordinator
- Reviewer (who is assigned the `respondent_subject_id` subject)

Each subject in the list returned will be the target subject of at least one:
- Assignment where `respondent_subject_id` is the respondent subject (including auto-assigned activities/flows)
- Submission where `respondent_subject_id` is the source subject

In addition to the usual `SubjectReadResponse` properties, there are two extra ones that I'd like to highlight:
- `submissionCount`: This is the number of answers that the respondent has submitted for a given subject and activity/flow
- `currentlyAssigned`: Whether the activity/flow is currently assigned for the respondent to complete about a given subject

Activities/flows that have been unassigned and don't have any submissions aren't included

src/apps/activity_assignments/crud/assignments.py

@@ -111,6 +111,44 @@ async def already_exists(self, schema: ActivityAssigmentSchema) -> ActivityAssig
         db_result = await self._execute(query)
         return db_result.scalars().first()
 
+    async def get_target_subject_ids_by_activity_or_flow_ids(
+        self,
+        respondent_subject_id: uuid.UUID,
+        activity_or_flow_ids: list[uuid.UUID] = [],
+    ) -> list[uuid.UUID]:
+        """
+        Retrieves the IDs of target subjects that have assignments to be completed by the provided respondent.
+
+        Parameters:
+        ----------
+        respondent_subject_id : uuid.UUID
+            The ID of the respondent subject to search for. This parameter is required.
+        activity_or_flow_ids : list[uuid.UUID]
+            Optional list of activity or flow IDs to narrow the search. These IDs may correspond to either
+            `activity_id` or `activity_flow_id` fields
+
+        Returns:
+        -------
+        list[uuid.UUID]
+            List of target subject IDs associated with the provided activity or flow IDs.
+        """
+        query = select(ActivityAssigmentSchema.target_subject_id).where(
+            ActivityAssigmentSchema.respondent_subject_id == respondent_subject_id,
+            ActivityAssigmentSchema.soft_exists(),
+        )
+
+        if len(activity_or_flow_ids) > 0:
+            query = query.where(
+                or_(
+                    ActivityAssigmentSchema.activity_id.in_(activity_or_flow_ids),
+                    ActivityAssigmentSchema.activity_flow_id.in_(activity_or_flow_ids),
+                )
+            )
+
+        db_result = await self._execute(query.distinct())
+
+        return db_result.scalars().all()
+
     async def delete_by_activity_or_flow_ids(self, activity_or_flow_ids: list[uuid.UUID]):
         """
         Marks the `is_deleted` field as True for all matching assignments based on the provided
@@ -263,3 +301,15 @@ async def upsert(self, values: dict) -> ActivityAssigmentSchema | None:
             updated_schema = await self._get("id", model_id)
 
         return updated_schema
+
+    async def check_if_auto_assigned(self, activity_or_flow_id: uuid.UUID) -> bool | None:
+        """
+        Checks if the activity or flow is currently set to auto-assign.
+        """
+        activities_query = select(ActivitySchema.auto_assign).where(ActivitySchema.id == activity_or_flow_id)
+        flows_query = select(ActivityFlowSchema.auto_assign).where(ActivityFlowSchema.id == activity_or_flow_id)
+
+        union_query = activities_query.union_all(flows_query).limit(1)
+
+        db_result = await self._execute(union_query)
+        return db_result.scalar_one_or_none()

src/apps/activity_assignments/service.py

@@ -399,6 +399,37 @@ async def get_all_with_subject_entities(
             for assignment in assignments
         ]
 
+    async def get_target_subject_ids_by_respondent(
+        self,
+        respondent_subject_id: uuid.UUID,
+        activity_or_flow_ids: list[uuid.UUID] = [],
+    ) -> list[uuid.UUID]:
+        """
+        Retrieves the IDs of target subjects that have assignments to be completed by the provided respondent.
+
+        Parameters:
+        ----------
+        respondent_subject_id : uuid.UUID
+            The ID of the respondent subject to search for. This parameter is required.
+        activity_or_flow_ids : list[uuid.UUID]
+            Optional list of activity or flow IDs to narrow the search. These IDs may correspond to either
+            `activity_id` or `activity_flow_id` fields
+
+        Returns:
+        -------
+        list[uuid.UUID]
+            List of target subject IDs associated with the provided activity or flow IDs.
+        """
+        return await ActivityAssigmentCRUD(self.session).get_target_subject_ids_by_activity_or_flow_ids(
+            respondent_subject_id, activity_or_flow_ids
+        )
+
+    async def check_if_auto_assigned(self, activity_or_flow_id: uuid.UUID) -> bool | None:
+        """
+        Checks if the activity or flow is currently set to auto-assign.
+        """
+        return await ActivityAssigmentCRUD(self.session).check_if_auto_assigned(activity_or_flow_id)
+
     @staticmethod
     def _get_email_template_name(language: str) -> str:
         return f"new_activity_assignments_{language}"

src/apps/answers/crud/answers.py

@@ -937,3 +937,24 @@ async def delete_by_ids(self, ids: list[uuid.UUID]):
         query: Query = delete(AnswerSchema)
         query = query.where(AnswerSchema.id.in_(ids))
         await self._execute(query)
+
+    async def get_target_subject_ids_by_respondent(
+        self, respondent_subject_id: uuid.UUID, activity_or_flow_id: uuid.UUID
+    ):
+        query: Query = (
+            select(
+                AnswerSchema.target_subject_id,
+                func.count(func.distinct(AnswerSchema.submit_id)).label("submission_count"),
+            )
+            .where(
+                AnswerSchema.source_subject_id == respondent_subject_id,
+                or_(
+                    AnswerSchema.id_from_history_id(AnswerSchema.activity_history_id) == str(activity_or_flow_id),
+                    AnswerSchema.id_from_history_id(AnswerSchema.flow_history_id) == str(activity_or_flow_id),
+                ),
+            )
+            .group_by(AnswerSchema.target_subject_id)
+        )
+
+        res = await self._execute(query)
+        return res.all()

src/apps/answers/service.py

@@ -1853,6 +1853,13 @@ async def _prepare_answer_reviews(
             )
         return results
 
+    async def get_target_subject_ids_by_respondent_and_activity_or_flow(
+        self, respondent_subject_id: uuid.UUID, activity_or_flow_id: uuid.UUID
+    ) -> list[tuple[uuid.UUID, int]]:
+        return await AnswersCRUD(self.answer_session).get_target_subject_ids_by_respondent(
+            respondent_subject_id, activity_or_flow_id
+        )
+
 
 class ReportServerService:
     def __init__(self, session, arbitrary_session=None):

src/apps/subjects/api.py

@@ -1,18 +1,20 @@
 import uuid
 from datetime import datetime, timedelta
+from typing import TypedDict
 
 from fastapi import Body, Depends
 from fastapi.exceptions import RequestValidationError
 from pydantic.error_wrappers import ErrorWrapper
 from sqlalchemy.ext.asyncio import AsyncSession
 
+from apps.activity_assignments.service import ActivityAssignmentService
 from apps.answers.deps.preprocess_arbitrary import get_answer_session, get_answer_session_by_subject
 from apps.answers.service import AnswerService
 from apps.applets.service import AppletService
 from apps.authentication.deps import get_current_user
 from apps.invitations.errors import NonUniqueValue
 from apps.invitations.services import InvitationsService
-from apps.shared.domain import Response
+from apps.shared.domain import Response, ResponseMulti
 from apps.shared.exception import NotFoundError, ValidationError
 from apps.shared.response import EmptyResponse
 from apps.shared.subjects import is_take_now_relation, is_valid_take_now_relation
@@ -24,6 +26,7 @@
     SubjectReadResponse,
     SubjectRelationCreate,
     SubjectUpdateRequest,
+    TargetSubjectByRespondentResponse,
 )
 from apps.subjects.errors import SecretIDUniqueViolationError
 from apps.subjects.services import SubjectsService
@@ -296,3 +299,84 @@ async def get_my_subject(
             last_name=subject.last_name,
         )
     )
+
+
+async def get_target_subjects_by_respondent(
+    respondent_subject_id: uuid.UUID,
+    activity_or_flow_id: uuid.UUID,
+    user: User = Depends(get_current_user),
+    session: AsyncSession = Depends(get_session),
+    answer_session=Depends(get_answer_session),
+) -> ResponseMulti[TargetSubjectByRespondentResponse]:
+    subjects_service = SubjectsService(session, user.id)
+    respondent_subject = await subjects_service.get(respondent_subject_id)
+    if not respondent_subject:
+        raise NotFoundError(f"Subject with id {respondent_subject_id} not found")
+
+    if respondent_subject.user_id is None:
+        # Return a generic bad request error to avoid leaking information
+        raise ValidationError(f"Subject {respondent_subject_id} is not a valid respondent")
+
+    # Make sure the authenticated user has access to the subject
+    await CheckAccessService(session, user.id).check_subject_subject_access(
+        respondent_subject.applet_id, respondent_subject.id
+    )
+
+    assignment_service = ActivityAssignmentService(session)
+    assignment_subject_ids = await assignment_service.get_target_subject_ids_by_respondent(
+        respondent_subject_id=respondent_subject_id, activity_or_flow_ids=[activity_or_flow_id]
+    )
+
+    is_auto_assigned = await assignment_service.check_if_auto_assigned(activity_or_flow_id)
+    if is_auto_assigned:
+        assignment_subject_ids.append(respondent_subject_id)
+
+    submission_data: list[tuple[uuid.UUID, int]] = await AnswerService(
+        user_id=user.id, session=session, arbitrary_session=answer_session
+    ).get_target_subject_ids_by_respondent_and_activity_or_flow(
+        respondent_subject_id=respondent_subject_id, activity_or_flow_id=activity_or_flow_id
+    )
+
+    class SubjectInfo(TypedDict):
+        currently_assigned: bool
+        submission_count: int
+
+    subject_info: dict[uuid.UUID, SubjectInfo] = {}
+    for subject_id, submission_count in submission_data:
+        subject_info[subject_id] = {"currently_assigned": False, "submission_count": submission_count}
+
+    for subject_id in assignment_subject_ids:
+        if subject_id not in subject_info:
+            subject_info[subject_id] = {"currently_assigned": True, "submission_count": 0}
+        else:
+            subject_info[subject_id]["currently_assigned"] = True
+
+    subjects: list[Subject] = await subjects_service.get_by_ids(list(subject_info.keys()))
+    result: list[TargetSubjectByRespondentResponse] = []
+
+    # Find the respondent subject in the list of subjects
+    respondent_target_subject: TargetSubjectByRespondentResponse | None = None
+    for subject in subjects:
+        target_subject = TargetSubjectByRespondentResponse(
+            secret_user_id=subject.secret_user_id,
+            nickname=subject.nickname,
+            tag=subject.tag,
+            id=subject.id,
+            applet_id=subject.applet_id,
+            user_id=subject.user_id,
+            first_name=subject.first_name,
+            last_name=subject.last_name,
+            submission_count=subject_info[subject.id]["submission_count"],
+            currently_assigned=subject_info[subject.id]["currently_assigned"],
+        )
+
+        if subject.id == respondent_subject_id:
+            respondent_target_subject = target_subject
+        else:
+            result.append(target_subject)
+
+    if respondent_target_subject:
+        # TODO: If this endpoint is ever paginated, this logic will need to be moved to the query level
+        result.insert(0, respondent_target_subject)
+
+    return ResponseMulti(result=result, count=len(result))

src/apps/subjects/domain.py

@@ -88,6 +88,11 @@ class SubjectReadResponse(SubjectUpdateRequest):
     last_name: str
 
 
+class TargetSubjectByRespondentResponse(SubjectReadResponse):
+    submission_count: int = 0
+    currently_assigned: bool = False
+
+
 class SubjectRelation(InternalModel):
     source_subject_id: uuid.UUID
     target_subject_id: uuid.UUID

src/apps/subjects/router.py

@@ -4,17 +4,25 @@
 from starlette import status
 
 from apps.authentication.deps import get_current_user
-from apps.shared.domain import AUTHENTICATION_ERROR_RESPONSES, DEFAULT_OPENAPI_RESPONSE, Response
+from apps.shared.domain import AUTHENTICATION_ERROR_RESPONSES, DEFAULT_OPENAPI_RESPONSE, Response, ResponseMulti
 from apps.subjects.api import (
     create_relation,
     create_subject,
     create_temporary_multiinformant_relation,
     delete_relation,
     delete_subject,
     get_subject,
+    get_target_subjects_by_respondent,
     update_subject,
 )
-from apps.subjects.domain import Subject, SubjectCreateRequest, SubjectCreateResponse, SubjectFull, SubjectReadResponse
+from apps.subjects.domain import (
+    Subject,
+    SubjectCreateRequest,
+    SubjectCreateResponse,
+    SubjectFull,
+    SubjectReadResponse,
+    TargetSubjectByRespondentResponse,
+)
 from apps.users import User
 from infrastructure.database.deps import get_session
 
@@ -104,3 +112,14 @@ async def create_shell_account(
         **AUTHENTICATION_ERROR_RESPONSES,
     },
 )(delete_relation)
+
+router.get(
+    "/respondent/{respondent_subject_id}/activity-or-flow/{activity_or_flow_id}",
+    response_model=ResponseMulti[TargetSubjectByRespondentResponse],
+    status_code=status.HTTP_200_OK,
+    responses={
+        status.HTTP_200_OK: {"model": ResponseMulti[TargetSubjectByRespondentResponse]},
+        **DEFAULT_OPENAPI_RESPONSE,
+        **AUTHENTICATION_ERROR_RESPONSES,
+    },
+)(get_target_subjects_by_respondent)

src/apps/subjects/services/subjects.py

@@ -69,6 +69,10 @@ async def get(self, id_: uuid.UUID) -> Subject | None:
         schema = await SubjectsCrud(self.session).get_by_id(id_)
         return Subject.from_orm(schema) if schema else None
 
+    async def get_by_ids(self, ids: list[uuid.UUID], include_deleted=False) -> list[Subject]:
+        subjects = await SubjectsCrud(self.session).get_by_ids(ids, include_deleted)
+        return [Subject.from_orm(subject) for subject in subjects]
+
     async def get_if_soft_exist(self, id_: uuid.UUID) -> Subject | None:
         schema = await SubjectsCrud(self.session).get_by_id(id_)
         if schema and schema.soft_exists():